Build connections

1. Run a working MCP connection

This project implements a real MCP server over stdio. Its search_logs tool queries a read-only SQLite connection, restricts access to one service, validates inputs, and redacts common secret fields. A separate client tests the actual MCP protocol.

Requirements: Python 3.11+, pip, and unzip. Download the complete MCP projectinto a new directory, then run:

unzip logs-mcp.zip
cd logs-mcp
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python make_logs.py
python check_connection.py

Expected output: PASS: discovery, search, redaction, scope denial, input validation. This uses local test records and makes no model requests.

2. Implement the tool

"""Read-only MCP log search. Requires Python 3.11+ and mcp==2.3.0."""
import os
import re
import sqlite3
from pathlib import Path
from typing import Annotated

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from pydantic import Field

mcp = MCPServer("Service logs")
SERVICE = os.environ["ALLOWED_SERVICE"]
DATABASE = Path(os.environ["LOG_DB_PATH"]).resolve().as_uri() + "?mode=ro"
SECRET = re.compile(r"(?i)\b(token|password|api[_-]?key|authorization)\s*[:=]\s*(?:bearer\s+)?\S+")


@mcp.tool()
def search_logs(
    service: str,
    query: Annotated[str, Field(min_length=1, max_length=500)],
    limit: Annotated[int, Field(ge=1, le=100)] = 20,
) -> list[dict[str, str]]:
    """Search the service authorized by this connection. Returns newest first."""
    if service != SERVICE:
        raise ToolError("Service is outside this connection's scope")
    with sqlite3.connect(DATABASE, uri=True, timeout=5) as connection:
        connection.row_factory = sqlite3.Row
        rows = connection.execute(
            "SELECT timestamp, level, message FROM logs "
            "WHERE service = ? AND instr(lower(message), lower(?)) > 0 "
            "ORDER BY timestamp DESC LIMIT ?",
            (SERVICE, query, limit),
        ).fetchall()
    return [
        {"timestamp": row["timestamp"], "level": row["level"],
         "message": SECRET.sub(r"\1=[REDACTED]", row["message"])[:2000]}
        for row in rows
    ]


if __name__ == "__main__":
    mcp.run(transport="stdio")

The connection runner supplies ALLOWED_SERVICE and LOG_DB_PATH from trusted configuration. The database opens withmode=ro, and the query uses bound parameters. Requests for a different service are denied before the database is read. Results are capped at 100 rows and 2,000 characters per message.

The included redaction targets common key-value secrets. Before connecting production logs, apply your data classification and redaction rules at the source as well.

3. Test the connection end to end

"""Exercise the actual MCP protocol, including denied and invalid calls."""
import asyncio
import json
import sys
from pathlib import Path

from mcp import Client, StdioServerParameters


async def main() -> None:
    directory = Path(__file__).resolve().parent
    server = StdioServerParameters(
        command=sys.executable,
        args=[str(directory / "logs_server.py")],
        env={"ALLOWED_SERVICE": "service-a", "LOG_DB_PATH": str(directory / "logs.sqlite")},
    )
    async with Client(server) as client:
        tools = await client.list_tools()
        assert any(tool.name == "search_logs" for tool in tools.tools)
        result = await client.call_tool(
            "search_logs", {"service": "service-a", "query": "API", "limit": 10}
        )
        assert not result.is_error, result.content
        output = json.dumps([item.model_dump(mode="json") for item in result.content])
        assert "API timeout" in output and "[REDACTED]" in output
        assert "private-test-token" not in output and "Restricted service" not in output
        denied = await client.call_tool(
            "search_logs", {"service": "service-b", "query": "error"}
        )
        assert denied.is_error, "An out-of-scope service must be denied"
        invalid = await client.call_tool(
            "search_logs", {"service": "service-a", "query": "API", "limit": 101}
        )
        assert invalid.is_error, "Limits above 100 must be rejected"
    print("PASS: discovery, search, redaction, scope denial, input validation")


if __name__ == "__main__":
    asyncio.run(main())

The test discovers the tool, searches Service A, checks redaction, and verifies that Service B and an excessive result limit are rejected. The test database deliberately includes a restricted record and a fake token so those checks can fail meaningfully.

4. Register through UI, CLI, or Terraform

Register the connection using the interface your team uses to manage configuration. Give it a name, a Capability Owner, its process command or authenticated endpoint, and an explicit scope. Keep secrets in connection configuration, outside the agent’s instructions and results.

For this stdio server, the process command is the absolute path to the virtual environment’s Python executable; its argument is the absolute path to logs_server.py. Configure ALLOWED_SERVICE as service-a and LOG_DB_PATH as the absolute path to logs.sqlite. The connection runner must be able to access both files.

Use stdio when the runner launches the server process. For a shared remote service, deploy an authenticated Streamable HTTP endpoint and configure the connection to use that transport. See the official MCP Python SDK documentationfor transport and authorization configuration.

5. Authorize every call

Connection-level scoping is one boundary. The harness must also verify the session’s current Execution Identity, delegation grant, policy, and limits before every call. A shared service account does not authorize every agent to use it.

1. Validate input and resolve the current execution identity.
2. Check grant expiry, revocation, resource scope, and action.
3. Reserve time and budget, and obtain any required approval.
4. Invoke the tool with the permitted connection.
5. Return minimal, redacted output with evidence references.
6. Record the decision, outcome, and actual usage without secrets.

Treat retrieved content as untrusted data. When adapting this project to an external API, keep its business rules and access controls in the source system, add timeouts, and test credential revocation. Writes also need idempotency and retry tests.