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#
News pipeline and Django site for the corpus an operator-tools server typically wraps.
Frameworks for the surrounding AI / agent context.
MCP Inspector, the protocol debugger.