Model Context Protocol#

A Python implementation of an MCP server that exposes the operator’s tools, resources, and prompts to any MCP-aware LLM client (Claude Desktop, Claude Code, IDE plugins, custom agents). Write an MCP server once; every MCP client can drive it.

MCP is an open protocol from Anthropic that standardises how LLMs discover and call tools, read context resources, and request prompts. The wire format is JSON-RPC 2.0; transports are stdio (subprocess) and streamable-http (remote).

Architecture#

Three roles. The host is the application the human uses (Claude Desktop, Claude Code, an agent). The host runs one or more clients, each talking to one server that exposes tools (callable functions), resources (read-only data), and prompts (reusable prompt templates).

        flowchart LR
    Human([Operator]) --> Host
    subgraph Host[Host application]
        direction TB
        LLM["LLM<br/>(Claude / agent)"]
        C1[MCP client A]
        C2[MCP client B]
        LLM <--> C1
        LLM <--> C2
    end
    C1 -->|JSON-RPC<br/>stdio| S1["MCP server<br/>(operator-tools)"]
    C2 -->|JSON-RPC<br/>HTTP| S2["MCP server<br/>(news-corpus)"]
    S1 --> T1[Tools]
    S1 --> R1[Resources]
    S2 --> R2[Resources]
    S2 --> P2[Prompts]
    

Write one server per capability bundle. A news-corpus server might expose search_articles (tool), article_by_id (resource), and a news_brief prompt template; a separate scan server exposes nmap_scan (tool) and last_scan_summary (resource).

Primitives#

Primitive

Direction

Operator concern

Tool

Client invokes server

Side-effectful actions the LLM can call (scan, query, compute). Defines a JSON-schema input.

Resource

Client reads from server

Read-only data the LLM can pull into its context (files, DB rows, search results). Identified by URI.

Prompt

Client requests from server

Reusable prompt templates the LLM can apply (with arguments).

Sampling

Server requests from client

The server asks the host’s LLM to generate completions (advanced; lets servers reason).

Wire flow#

Initialise, advertise capabilities, then call. Sketch of an tools/call for the search_articles tool.

        sequenceDiagram
    participant H as Host (LLM)
    participant C as MCP client
    participant S as MCP server

    C->>S: initialize {protocolVersion, capabilities}
    S-->>C: initialize result {serverInfo, capabilities}
    C->>S: notifications/initialized
    C->>S: tools/list
    S-->>C: tools [search_articles, ...]
    H->>C: I want to call search_articles
    C->>S: tools/call {name: "search_articles", arguments: {...}}
    S-->>C: content [{type: "text", text: "..."}]
    H-->>H: incorporate results into next turn
    

The whole exchange is JSON-RPC 2.0; over stdio each side reads and writes newline-delimited JSON to file descriptors; over HTTP, streamable-http uses HTTP requests with SSE for server-to- client notifications.

Project layout#

mcp-operator-tools/
├── pyproject.toml
├── README.md
└── src/
    └── operator_tools/
        ├── __init__.py
        ├── server.py            # MCP wiring
        ├── tools/
        │   ├── search.py        # search_articles tool
        │   └── scan.py          # run_scan tool
        ├── resources/
        │   └── article.py       # article resource handler
        └── prompts/
            └── brief.py         # news_brief prompt

Server skeleton#

Using the official mcp Python SDK (modelcontextprotocol/python-sdk).

$ uv add mcp
# src/operator_tools/server.py
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("operator-tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_articles",
            description="Search the news corpus by query and date range.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query":  {"type": "string"},
                    "since":  {"type": "string", "format": "date"},
                    "limit":  {"type": "integer", "default": 10},
                },
                "required": ["query"],
            },
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_articles":
        from .tools.search import search_articles
        results = await search_articles(**arguments)
        return [TextContent(type="text", text=str(results))]
    raise ValueError(f"unknown tool: {name}")

async def main() -> None:
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

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

A tool handler.

# src/operator_tools/tools/search.py
import httpx

async def search_articles(query: str, since: str | None = None,
                          limit: int = 10) -> list[dict]:
    params = {"q": query, "since": since, "limit": limit}
    async with httpx.AsyncClient() as client:
        r = await client.get("http://news-site.local/api/search",
                             params=params)
        r.raise_for_status()
        return r.json()

A resource handler.

# src/operator_tools/resources/article.py
from mcp.server import Server
from mcp.types import Resource, ResourceContents, TextResourceContents

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="article://corpus/{id}",
            name="Article",
            mimeType="text/plain",
            description="Full text of an article by id.",
        ),
    ]

@server.read_resource()
async def read_resource(uri: str) -> ResourceContents:
    art_id = uri.removeprefix("article://corpus/")
    text = fetch_article_text(int(art_id))
    return TextResourceContents(uri=uri, mimeType="text/plain",
                                text=text)

A prompt template.

# src/operator_tools/prompts/brief.py
from mcp.types import Prompt, PromptArgument, GetPromptResult, PromptMessage

@server.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="news_brief",
            description="Summarize corpus search results into a brief.",
            arguments=[
                PromptArgument(name="topic", required=True),
                PromptArgument(name="window_days", required=False),
            ],
        ),
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
    return GetPromptResult(
        messages=[
            PromptMessage(
                role="user",
                content={
                    "type": "text",
                    "text": (
                        f"Write a one-page intelligence brief on "
                        f"{arguments['topic']}, last "
                        f"{arguments.get('window_days', 30)} days. "
                        f"Source from the corpus. Cite article IDs."
                    ),
                },
            ),
        ],
    )

Wire it to a host#

Claude Desktop and Claude Code both read MCP server configuration from a JSON file. Example for Claude Code (project-scoped).

{
  "mcpServers": {
    "operator-tools": {
      "command": "uv",
      "args": ["run", "python", "-m", "operator_tools.server"],
      "cwd": "/home/operator/mcp-operator-tools"
    }
  }
}

After registering, the host’s LLM sees search_articles, article://corpus/{id}, and news_brief automatically. The operator drives the tools through the LLM.

Common Tasks#

Scaffold a new MCP server project.

$ uv init mcp-operator-tools && cd mcp-operator-tools
$ uv add mcp httpx

Run the server locally and test with the MCP inspector.

$ uv run python -m operator_tools.server
# in another shell
$ npx @modelcontextprotocol/inspector \
    uv run python -m operator_tools.server

Smoke-test a tool via JSON-RPC by hand (stdio).

$ printf '%s\n' '{"jsonrpc":"2.0","id":1,"method":"initialize",\
  "params":{"protocolVersion":"2025-06-18","capabilities":{},\
  "clientInfo":{"name":"shell","version":"0"}}}' | \
    uv run python -m operator_tools.server

Register the server with Claude Code.

$ claude mcp add operator-tools \
    --command "uv run python -m operator_tools.server" \
    --cwd "$PWD"

Serve over HTTP instead of stdio.

# server.py
from mcp.server.streamable_http import streamable_http_server
# ...replace stdio_server() with streamable_http_server(host, port)

References#