refactor: replace lowlevel Server decorators with on_* constructor kwargs

[maxisbey]

Replace the decorator-based handler registration on the lowlevel Server with direct on_* keyword arguments on the constructor. Handlers are now raw callables with a uniform (ctx, params) -> result signature, dispatched by method string.

Supersedes #1968.

[claude]

Code review

No issues found. Checked for bugs and CLAUDE.md compliance.

[maxisbey]

Although CI is all green, I still want to manually test all the examples that were modified.

[Kludex]

Please check if copilot's comments make sense. I don't think it's reasonable for me to review at this point - but if there's anything specific that you want me to look at, happy to.

[Copilot]

Pull request overview

This PR completes the refactor of the low-level Server API by removing decorator-based handler registration and moving to constructor on_* keyword arguments with uniform (ctx, params) -> result handler signatures. It updates the test suite, examples, and supporting server/context utilities to align with the new handler model and method-string dispatch.

Changes:

• Migrates tests and examples from @server.*() decorators to Server(..., on_*=...) handlers returning concrete MCP result models.
• Updates server/context plumbing to support the new request context flow (context passed into handlers; request context fields now optional for notifications).
• Refreshes documentation/migration guidance to describe the new low-level Server API and handler contracts.

Reviewed changes

Copilot reviewed 71 out of 71 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
tests/shared/test_ws.py	Updates WS test server to use on_* constructor handlers and typed params/results.
tests/shared/test_sse.py	Updates SSE test server(s) to on_* handlers; shifts request-context usage to ctx.
tests/shared/test_session.py	Updates cancellation test to new ServerRequestContext + CallToolRequestParams signature.
tests/shared/test_progress_notifications.py	Updates progress notification handlers and tool calls to new meta/params model.
tests/shared/test_memory.py	Removes legacy memory-stream example test that depended on decorator API.
tests/server/test_streamable_http_manager.py	Updates streamable-http tests to new low-level Server imports and on_list_tools.
tests/server/test_session.py	Updates capability tests to rely on on_* presence instead of decorators.
tests/server/test_read_resource.py	Reworks read-resource tests to use Client(server) and explicit ReadResourceResult contents.
tests/server/test_lowlevel_tool_annotations.py	Simplifies tool-annotation test using Client(server) and on_list_tools.
tests/server/test_lowlevel_output_validation.py	Removes legacy low-level output validation tests (old decorator/validation behavior).
tests/server/test_lowlevel_input_validation.py	Removes legacy low-level input validation tests (old decorator/validation behavior).
tests/server/test_lifespan.py	Updates lifespan/tool handler to new (ctx, params) signature and result type.
tests/server/test_completion_with_context.py	Updates completion tests to use on_completion and CompleteRequestParams/CompleteResult.
tests/server/test_cancel_handling.py	Updates cancel-handling test to use on_list_tools/on_call_tool and CallToolResult.
tests/server/mcpserver/test_server.py	Adds/updates MCPServer completion decorator test coverage.
tests/server/mcpserver/prompts/test_manager.py	Fixes imports to use TextContent from mcp.types.
tests/server/mcpserver/prompts/test_base.py	Fixes imports to use TextContent from mcp.types.
tests/server/mcpserver/auth/test_auth_integration.py	Updates auth settings imports split between routes/settings modules.
tests/server/lowlevel/test_server_pagination.py	Updates pagination tests to Client(server) flow and (ctx, params) handlers.
tests/server/lowlevel/test_server_listing.py	Updates list tests to Client(server) and explicit result models.
tests/server/lowlevel/test_func_inspection.py	Removes tests for old signature-inspection wrapper (decorator era).
tests/issues/test_88_random_error.py	Updates reproduction test to new low-level server handler signatures/results.
tests/issues/test_342_base64_encoding.py	Updates base64 regression test to use MCPServer + Client read_resource path.
tests/issues/test_1574_resource_uri_validation.py	Updates URI roundtrip tests to new low-level handler signatures/results.
tests/issues/test_152_resource_mime_type.py	Updates mime-type test to explicit TextResourceContents/BlobResourceContents results.
tests/issues/test_129_resource_templates.py	Updates resource-template listing test to use Client(mcp) rather than internal handler access.
tests/experimental/tasks/test_spec_compliance.py	Migrates to experimental.enable_tasks(on_*=...) and skips partial-capability tests for now.
tests/experimental/tasks/test_elicitation_scenarios.py	Migrates scenario servers to on_* handlers and uses ctx.session APIs.
tests/experimental/tasks/server/test_integration.py	Refactors integration tests to use Client(server) and lifespan context patterns.
tests/client/transports/test_memory.py	Updates in-memory transport test server to on_list_resources and ListResourcesResult.
tests/client/test_output_schema_validation.py	Refactors malicious-server simulation to just return invalid structured content from low-level server.
tests/client/test_list_methods_cursor.py	Updates low-level server cursor echo test to (ctx, params) list-tools handler.
tests/client/test_http_unicode.py	Migrates unicode server example in tests to on_* handlers and typed params/results.
tests/client/test_client.py	Updates client fixture server to provide required capabilities via on_* handlers returning EmptyResult/CompleteResult.
src/mcp/shared/message.py	Types request_context as Any to keep transport-agnostic metadata while allowing None.
src/mcp/shared/experimental/tasks/helpers.py	Updates docs/example snippets to new (ctx, params) task handlers.
src/mcp/shared/_context.py	Makes request_id optional so one context type can support both requests and notifications.
src/mcp/server/streamable_http_manager.py	Updates Server generic usage after type parameter simplification.
src/mcp/server/session.py	Updates module docs/example to the new handler API and explicit result types.
src/mcp/server/mcpserver/server.py	Wires core MCP protocol handlers via low-level Server(..., on_*=...) and updates context access.
src/mcp/server/lowlevel/func_inspection.py	Removes obsolete decorator-era call wrapper utility.
src/mcp/server/lowlevel/experimental.py	Replaces decorator registration with enable_tasks(on_*=...) and method-string handler wiring.
src/mcp/server/lowlevel/init.py	Minor __all__ ordering update.
src/mcp/server/experimental/task_result_handler.py	Updates usage docs to reflect enable_tasks(on_task_result=...).
src/mcp/server/experimental/request_context.py	Updates example snippet to the new handler signature.
src/mcp/server/context.py	Sets default typevar for lifespan context and keeps request type default as Any.
src/mcp/server/init.py	Exports ServerRequestContext at mcp.server top-level.
examples/snippets/servers/pagination_example.py	Migrates snippet to on_list_resources and (ctx, params) pagination.
examples/snippets/servers/lowlevel/structured_output.py	Migrates snippet to explicit CallToolResult (content + structured_content) and create_initialization_options().
examples/snippets/servers/lowlevel/lifespan.py	Migrates snippet to on_* handlers and typed lifespan context.
examples/snippets/servers/lowlevel/direct_call_tool_result.py	Migrates snippet to on_* handlers returning CallToolResult directly.
examples/snippets/servers/lowlevel/basic.py	Migrates snippet to on_list_prompts/on_get_prompt and explicit result types.
examples/servers/structured-output-lowlevel/mcp_structured_output_lowlevel/main.py	Migrates example to explicit CallToolResult and create_initialization_options().
examples/servers/sse-polling-demo/mcp_sse_polling_demo/server.py	Migrates example handlers to on_* and uses ctx.session for logs/stream control.
examples/servers/simple-tool/mcp_simple_tool/server.py	Migrates to on_list_tools/on_call_tool and returns CallToolResult.
examples/servers/simple-task/mcp_simple_task/server.py	Migrates to on_* handlers; task support enabled via experimental.enable_tasks().
examples/servers/simple-task-interactive/mcp_simple_task_interactive/server.py	Migrates to on_* handlers and passes ctx through to task helpers.
examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py	Migrates to on_* handlers and uses StreamableHTTPSessionManager with new server type.
examples/servers/simple-streamablehttp-stateless/mcp_simple_streamablehttp_stateless/server.py	Migrates stateless streamable-http example to on_* handlers.
examples/servers/simple-resource/mcp_simple_resource/server.py	Migrates resources example to on_* and explicit ReadResourceResult.
examples/servers/simple-prompt/mcp_simple_prompt/server.py	Migrates prompts example to on_* and typed params/results.
examples/servers/simple-pagination/mcp_simple_pagination/server.py	Refactors pagination example to shared helper + on_* handlers for all list/read/call methods.
examples/servers/everything-server/mcp_everything_server/server.py	Updates low-level handler registration via _add_request_handler and new request param types.
docs/migration.md	Adds migration guidance for on_* handlers, kw-only constructor params, and removed auto-wrapping.
docs/experimental/index.md	Updates tasks docs to emphasize experimental.enable_tasks() rather than decorators.
README.v2.md	Updates embedded snippets to match new low-level handler API and init options helper.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[maxisbey]

Although CI is all green, I still want to manually test all the examples that were modified.

According to Claude the examples pass, will try a few myself as well

[maxisbey]

Just for tracking, here is the testing scripts Claude wrote, as well as steps for running each of the examples one by one.

Script to run all examples files

```bash #!/bin/bash # Record asciinema demos showing real shell commands users can copy/paste. # Each recording shows: shell prompt → command → output set -e cd /fake/path/python-sdk mkdir -p evidence

Helper: simulate typing a command at a shell prompt, then run it

run_cmd() {
echo ""
echo -e "\033[1;32m$\033[0m $1"
sleep 0.3
eval "$1"
}

── 1. simple-tool ──────────────────────────────────────────────────────────

echo "Recording 01-simple-tool..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-tool example"
--command 'bash -c "
run_cmd() { echo ""; echo -e "\033[1;32m\$\033[0m $1"; sleep 0.3; eval "$1"; }

echo "━━━ simple-tool (stdio) ━━━"
echo "A server with a \"fetch\" tool that fetches website content."
echo ""

run_cmd "cd examples/servers/simple-tool"

cat << "SCRIPT"

$ uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\",
args=[\"run\", \"mcp-simple-tool\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
for t in tools.tools:
print(f\" Tool: {t.name}\")
print(f\" Description: {t.description}\")
anyio.run(main)
"
SCRIPT

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-tool\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
for t in tools.tools:
print(f\" Tool: {t.name}\")
print(f\" Description: {t.description}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/01-simple-tool.cast 2>&1 | tail -1

── 2. simple-prompt ────────────────────────────────────────────────────────

echo "Recording 02-simple-prompt..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-prompt example"
--command 'bash -c "
echo "━━━ simple-prompt (stdio) ━━━"
echo "A server with a \"simple\" prompt that takes optional context and topic."
echo ""

echo -e "\033[1;32m$\033[0m cd examples/servers/simple-prompt"
cd examples/servers/simple-prompt

cat << "SCRIPT"

$ uv run python -c "...client script..."
SCRIPT

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-prompt\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
prompts = await session.list_prompts()
for p in prompts.prompts:
print(f\" Prompt: {p.name} - {p.description}\")
print()
result = await session.get_prompt(\"simple\", {\"context\": \"testing kwargs\", \"topic\": \"MCP SDK\"})
for msg in result.messages:
print(f\" [{msg.role}]: {msg.content.text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/02-simple-prompt.cast 2>&1 | tail -1

── 3. simple-resource ──────────────────────────────────────────────────────

echo "Recording 03-simple-resource..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-resource example"
--command 'bash -c "
echo "━━━ simple-resource (stdio) ━━━"
echo "A server with 3 text resources: greeting, help, about."
echo ""
echo -e "\033[1;32m$\033[0m cd examples/servers/simple-resource"
cd examples/servers/simple-resource
echo -e "\033[1;32m$\033[0m uv run python -c \"...client...\""
echo ""
uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-resource\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
resources = await session.list_resources()
print(f\" Found {len(resources.resources)} resources:\")
for r in resources.resources:
result = await session.read_resource(r.uri)
print(f\" {r.name}: {result.contents[0].text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/03-simple-resource.cast 2>&1 | tail -1

── 4. simple-pagination ────────────────────────────────────────────────────

echo "Recording 04-simple-pagination..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-pagination example"
--command 'bash -c "
echo "━━━ simple-pagination (stdio) ━━━"
echo "25 tools (5/page), 30 resources (10/page), 20 prompts (7/page)."
echo ""
echo -e "\033[1;32m$\033[0m cd examples/servers/simple-pagination"
cd examples/servers/simple-pagination
echo -e "\033[1;32m$\033[0m uv run python -c \"...client...\""
echo ""
uv run python -c "
import anyio
from mcp import types
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-pagination\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
all_tools, cursor, page = [], None, 0
while True:
p = types.PaginatedRequestParams(cursor=cursor) if cursor else None
result = await session.list_tools(params=p)
all_tools.extend(result.tools)
page += 1
print(f\" Page {page}: {len(result.tools)} tools (next_cursor={result.next_cursor})\")
if not result.next_cursor: break
cursor = result.next_cursor
print(f\" Total: {len(all_tools)} tools across {page} pages\")
print()
result = await session.call_tool(\"tool_1\", {\"input\": \"hello\"})
print(f\" call_tool result: {result.content[0].text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/04-simple-pagination.cast 2>&1 | tail -1

── 5-8. Snippets ───────────────────────────────────────────────────────────

echo "Recording 05-snippet-basic..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: basic.py"
--command 'bash -c "
echo "━━━ snippet: lowlevel/basic.py (stdio) ━━━"
echo "Basic low-level server with prompts using kwargs constructor."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/basic.py"
echo " (server starts via stdio, client connects automatically)"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/basic.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
prompts = await session.list_prompts()
print(f\" Prompts: {[p.name for p in prompts.prompts]}\")
result = await session.get_prompt(\"example-prompt\", {\"arg1\": \"test-value\"})
print(f\" Message: {result.messages[0].content.text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/05-snippet-basic.cast 2>&1 | tail -1

echo "Recording 06-snippet-direct-call-tool..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: direct_call_tool_result.py"
--command 'bash -c "
echo "━━━ snippet: direct_call_tool_result.py (stdio) ━━━"
echo "Tool returning CallToolResult with structured_content and _meta."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/direct_call_tool_result.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/direct_call_tool_result.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
result = await session.call_tool(\"advanced_tool\", {\"message\": \"hello\"})
print(f\" Text: {result.content[0].text}\")
print(f\" Structured: {result.structured_content}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/06-snippet-direct-call-tool.cast 2>&1 | tail -1

echo "Recording 07-snippet-lifespan..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: lifespan.py"
--command 'bash -c "
echo "━━━ snippet: lifespan.py (stdio) ━━━"
echo "Server with lifespan: database connects on start, disconnects on stop."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/lifespan.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/lifespan.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
result = await session.call_tool(\"query_db\", {\"query\": \"SELECT * FROM users\"})
print(f\" Result: {result.content[0].text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/07-snippet-lifespan.cast 2>&1 | tail -1

echo "Recording 08-snippet-structured-output..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: structured_output.py"
--command 'bash -c "
echo "━━━ snippet: structured_output.py (stdio) ━━━"
echo "Tool with output_schema and structured_content response."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/structured_output.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/structured_output.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
print(f\" Tool: {tools.tools[0].name}\")
print(f\" Has output_schema: {tools.tools[0].output_schema is not None}\")
result = await session.call_tool(\"get_weather\", {\"city\": \"San Francisco\"})
print(f\" Structured: {result.structured_content}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/08-snippet-structured-output.cast 2>&1 | tail -1

── 9. structured-output-lowlevel ───────────────────────────────────────────

echo "Recording 09-structured-output-lowlevel..."
asciinema rec --overwrite --idle-time-limit 2 --title "structured-output-lowlevel example"
--command 'bash -c "
echo "━━━ structured-output-lowlevel (stdio) ━━━"
echo "Full server example with structured output (simulated weather)."
echo ""
echo -e "\033[1;32m$\033[0m python examples/servers/structured-output-lowlevel/mcp_structured_output_lowlevel/main.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\",
args=[\"run\", \"--frozen\", \"python\",
\"examples/servers/structured-output-lowlevel/mcp_structured_output_lowlevel/main.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
print(f\" Tool: {tools.tools[0].name} (output_schema: {tools.tools[0].output_schema is not None})\")
result = await session.call_tool(\"get_weather\", {\"city\": \"Tokyo\"})
print(f\" Structured: {result.structured_content}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/09-structured-output-lowlevel.cast 2>&1 | tail -1

── 10-15. HTTP servers ──────────────────────────────────────────────────────

record_http_demo() {
local num="$1"
local name="$2"
local title="$3"
local desc="$4"
local server_dir="$5"
local server_cmd="$6"
local port="$7"
local client_code="$8"

echo "Recording ${num}-${name}..."
asciinema rec --overwrite --idle-time-limit 2 --title "$title" \
  --command "bash -c \"

echo \"━━━ $title ━━━\"
echo \"$desc\"
echo \"\"
echo -e \"\\033[1;34m# Terminal 1: Start server\\033[0m\"
echo -e \"\\033[1;32m\\$\\033[0m cd $server_dir\"
echo -e \"\\033[1;32m\\$\\033[0m $server_cmd\"
echo \"\"

cd $server_dir
$server_cmd &
SERVER_PID=\$!
cd /fake/path/python-sdk

for i in \$(seq 1 30); do
code=\$(curl -s -o /dev/null -w \"%{http_code}\" http://127.0.0.1:$port/mcp 2>/dev/null || echo \"000\")
if echo \"\$code\" | grep -qE \"(200|307|400|405)\"; then break; fi
sleep 0.5
done

echo \" Server is running on http://127.0.0.1:$port\\\"
echo \"\"
echo -e \"\\033[1;34m# Terminal 2: Connect client\\033[0m\"
echo -e \"\\033[1;32m\\$\\033[0m uv run --frozen python -c \\\"...client...\\\"\"
echo \"\"

uv run --frozen python -c \"$client_code\"

echo \"\"
echo \"✓ Done\"

kill \$SERVER_PID 2>/dev/null; wait \$SERVER_PID 2>/dev/null
"" "evidence/${num}-${name}.cast" 2>&1 | tail -1
}

record_http_demo "10" "simple-streamablehttp"
"simple-streamablehttp (HTTP)"
"Streamable HTTP with notification stream and event store."
"examples/servers/simple-streamablehttp"
"uv run mcp-simple-streamablehttp --port 23000"
"23000"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23000/mcp') as client:
tools = await client.list_tools()
print(f' Tools: {[t.name for t in tools.tools]}')
result = await client.call_tool('start-notification-stream', {'interval': 0.1, 'count': 3, 'caller': 'demo'})
print(f' Result: {result.content[0].text}')

anyio.run(main)
"

record_http_demo "11" "simple-streamablehttp-stateless"
"simple-streamablehttp-stateless (HTTP)"
"Stateless streamable HTTP (no session persistence)."
"examples/servers/simple-streamablehttp-stateless"
"uv run mcp-simple-streamablehttp-stateless --port 23001"
"23001"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23001/mcp') as client:
tools = await client.list_tools()
print(f' Tools: {[t.name for t in tools.tools]}')
result = await client.call_tool('start-notification-stream', {'interval': 0.1, 'count': 3, 'caller': 'demo'})
print(f' Result: {result.content[0].text}')

anyio.run(main)
"

record_http_demo "12" "everything-server"
"everything-server (HTTP)"
"Conformance server: 13 tools, resources, prompts, completions."
"examples/servers/everything-server"
"uv run mcp-everything-server --port 23002"
"23002"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23002/mcp') as client:
tools = await client.list_tools()
print(f' Tools ({len(tools.tools)}): {[t.name for t in tools.tools]}')
result = await client.call_tool('test_simple_text', {})
print(f' test_simple_text: {result.content[0].text}')
resources = await client.list_resources()
print(f' Resources: {[str(r.uri) for r in resources.resources]}')
prompts = await client.list_prompts()
print(f' Prompts: {[p.name for p in prompts.prompts]}')

anyio.run(main)
"

record_http_demo "13" "simple-task"
"simple-task (HTTP)"
"Task server with long-running tool requiring task-augmented invocation."
"examples/servers/simple-task"
"uv run mcp-simple-task --port 23003"
"23003"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23003/mcp') as client:
tools = await client.list_tools()
t = tools.tools[0]
print(f' Tool: {t.name}')
print(f' task_support: {t.execution.task_support}')

anyio.run(main)
"

record_http_demo "14" "simple-task-interactive"
"simple-task-interactive (HTTP)"
"Interactive tasks: elicitation (confirm_delete) and sampling (write_haiku)."
"examples/servers/simple-task-interactive"
"uv run mcp-simple-task-interactive --port 23004"
"23004"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23004/mcp') as client:
tools = await client.list_tools()
for t in tools.tools:
print(f' Tool: {t.name} (task_support={t.execution.task_support})')

anyio.run(main)
"

record_http_demo "15" "sse-polling-demo"
"sse-polling-demo (HTTP)"
"SSE polling with periodic stream close and client auto-reconnect."
"examples/servers/sse-polling-demo"
"uv run mcp-sse-polling-demo --port 23005"
"23005"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23005/mcp') as client:
tools = await client.list_tools()
print(f' Tools: {[t.name for t in tools.tools]}')
result = await client.call_tool('process_batch', {'items': 3, 'checkpoint_every': 10})
print(f' Result: {result.content[0].text}')

anyio.run(main)
"

echo ""
echo "=========================================="
echo "All recordings complete!"
echo "=========================================="
echo ""
echo "Converting to GIFs..."
for cast in evidence/[0-9].cast; do
gif="${cast%.cast}.gif"
name=$(basename "$cast" .cast)
echo -n " $name -> gif... "
agg --idle-time-limit 1.5 --speed 1.5 "$cast" "$gif" 2>&1 | tail -1
done
echo ""
echo "All done!"
ls -lh evidence/[0-9].gif

<details>

<details>
<summary>Steps to run all 15 examples</summary>

All commands assume you are in the repository root directory.

---

### 1. simple-tool (stdio)

**What it does:** Provides a `fetch` tool that fetches website content.

```bash
cd examples/servers/simple-tool

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-tool', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            tools = await session.list_tools()
            for t in tools.tools:
                print(f'Tool: {t.name} - {t.description}')

anyio.run(main)
"

Expected output:

Tool: fetch - Fetches a website and returns its content

---

2. simple-prompt (stdio)

What it does: Provides a simple prompt with optional context and topic arguments.

cd examples/servers/simple-prompt

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-prompt', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            prompts = await session.list_prompts()
            for p in prompts.prompts:
                print(f'Prompt: {p.name} - {p.description}')
            result = await session.get_prompt('simple', {'context': 'hello', 'topic': 'MCP'})
            for msg in result.messages:
                print(f'  [{msg.role}]: {msg.content.text}')

anyio.run(main)
"

Expected output:

Prompt: simple - A simple prompt that can take optional context and topic arguments
  [user]: Here is some relevant context: hello
  [user]: Please help me with the following topic: MCP

---

3. simple-resource (stdio)

What it does: Provides 3 text resources: greeting, help, about.

cd examples/servers/simple-resource

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-resource', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            resources = await session.list_resources()
            for r in resources.resources:
                result = await session.read_resource(r.uri)
                print(f'{r.name}: {result.contents[0].text}')

anyio.run(main)
"

Expected output:

greeting: Hello! This is a sample text resource.
help: This server provides a few sample text resources for testing.
about: This is the simple-resource MCP server implementation.

---

4. simple-pagination (stdio)

What it does: Demonstrates pagination with 25 tools (5/page), 30 resources (10/page), 20 prompts (7/page).

cd examples/servers/simple-pagination

uv run python -c "
import anyio
from mcp import types
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-pagination', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            all_tools, cursor, page = [], None, 0
            while True:
                p = types.PaginatedRequestParams(cursor=cursor) if cursor else None
                result = await session.list_tools(params=p)
                all_tools.extend(result.tools)
                page += 1
                print(f'Page {page}: {len(result.tools)} tools')
                if not result.next_cursor: break
                cursor = result.next_cursor
            print(f'Total: {len(all_tools)} tools across {page} pages')

anyio.run(main)
"

Expected output:

Page 1: 5 tools
Page 2: 5 tools
Page 3: 5 tools
Page 4: 5 tools
Page 5: 5 tools
Total: 25 tools across 5 pages

---

5. snippet: lowlevel/basic.py (stdio)

What it does: Basic low-level server with prompts using the kwargs constructor pattern.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/basic.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            prompts = await session.list_prompts()
            print(f'Prompts: {[p.name for p in prompts.prompts]}')
            result = await session.get_prompt('example-prompt', {'arg1': 'test'})
            print(f'Message: {result.messages[0].content.text}')

anyio.run(main)
"

Expected output:

Prompts: ['example-prompt']
Message: Example prompt text with argument: test

---

6. snippet: lowlevel/direct_call_tool_result.py (stdio)

What it does: Tool returning CallToolResult directly with structured_content and _meta.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/direct_call_tool_result.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            result = await session.call_tool('advanced_tool', {'message': 'hello'})
            print(f'Text: {result.content[0].text}')
            print(f'Structured: {result.structured_content}')

anyio.run(main)
"

Expected output:

Text: Processed: hello
Structured: {'result': 'success', 'message': 'hello'}

---

7. snippet: lowlevel/lifespan.py (stdio)

What it does: Server with lifespan management — connects a mock database on startup, disconnects on shutdown.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/lifespan.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            result = await session.call_tool('query_db', {'query': 'SELECT * FROM users'})
            print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output:

Result: Query results: [{'id': '1', 'name': 'Example', 'query': 'SELECT * FROM users'}]

---

8. snippet: lowlevel/structured_output.py (stdio)

What it does: Tool with output_schema and structured_content response.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/structured_output.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            tools = await session.list_tools()
            print(f'Has output_schema: {tools.tools[0].output_schema is not None}')
            result = await session.call_tool('get_weather', {'city': 'San Francisco'})
            print(f'Structured: {result.structured_content}')

anyio.run(main)
"

Expected output:

Has output_schema: True
Structured: {'temperature': 22.5, 'condition': 'partly cloudy', 'humidity': 65, 'city': 'San Francisco'}

---

9. structured-output-lowlevel (stdio)

What it does: Full example server with structured output (simulated weather data).

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv',
        args=['run', '--frozen', 'python',
              'examples/servers/structured-output-lowlevel/mcp_structured_output_lowlevel/__main__.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            result = await session.call_tool('get_weather', {'city': 'Tokyo'})
            print(f'Structured keys: {list(result.structured_content.keys())}')

anyio.run(main)
"

Expected output:

Structured keys: ['temperature', 'conditions', 'humidity', 'wind_speed', 'timestamp']

---

10. simple-streamablehttp (HTTP)

What it does: Streamable HTTP server with notification stream tool and event store for resumability.

Terminal 1 — Start the server:

cd examples/servers/simple-streamablehttp
uv run mcp-simple-streamablehttp --port 3000

Wait until you see: Uvicorn running on http://127.0.0.1:3000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3000/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools: {[t.name for t in tools.tools]}')
        result = await client.call_tool('start-notification-stream', {'interval': 0.5, 'count': 3, 'caller': 'test'})
        print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output:

Tools: ['start-notification-stream']
Result: Sent 3 notifications with 0.5s interval for caller: test

---

11. simple-streamablehttp-stateless (HTTP)

What it does: Stateless streamable HTTP server (no session persistence between requests).

Terminal 1 — Start the server:

cd examples/servers/simple-streamablehttp-stateless
uv run mcp-simple-streamablehttp-stateless --port 3000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3000/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools: {[t.name for t in tools.tools]}')
        result = await client.call_tool('start-notification-stream', {'interval': 0.5, 'count': 3, 'caller': 'test'})
        print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output: Same as #10.

---

12. everything-server (HTTP)

What it does: Conformance test server implementing all MCP features — 13 tools, resources, prompts, completions, subscriptions, sampling, elicitation.

Terminal 1 — Start the server:

cd examples/servers/everything-server
uv run mcp-everything-server --port 3001

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3001/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools ({len(tools.tools)}): {[t.name for t in tools.tools]}')
        result = await client.call_tool('test_simple_text', {})
        print(f'test_simple_text: {result.content[0].text}')
        resources = await client.list_resources()
        print(f'Resources: {[str(r.uri) for r in resources.resources]}')
        prompts = await client.list_prompts()
        print(f'Prompts: {[p.name for p in prompts.prompts]}')

anyio.run(main)
"

Expected output:

Tools (13): ['test_simple_text', 'test_image_content', 'test_audio_content', ...]
test_simple_text: This is a simple text response for testing.
Resources: ['test://static-text', 'test://static-binary', 'test://watched-resource']
Prompts: ['test_simple_prompt', 'test_prompt_with_arguments', 'test_prompt_with_embedded_resource', 'test_prompt_with_image']

---

13. simple-task (HTTP)

What it does: Demonstrates MCP tasks with a long-running tool that requires task-augmented invocation.

Terminal 1 — Start the server:

cd examples/servers/simple-task
uv run mcp-simple-task --port 8000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:8000/mcp') as client:
        tools = await client.list_tools()
        t = tools.tools[0]
        print(f'Tool: {t.name}')
        print(f'  task_support: {t.execution.task_support}')

anyio.run(main)
"

Expected output:

Tool: long_running_task
  task_support: required

---

14. simple-task-interactive (HTTP)

What it does: Interactive task server demonstrating elicitation and sampling via tasks.

Terminal 1 — Start the server:

cd examples/servers/simple-task-interactive
uv run mcp-simple-task-interactive --port 8000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:8000/mcp') as client:
        tools = await client.list_tools()
        for t in tools.tools:
            print(f'Tool: {t.name} (task_support={t.execution.task_support})')

anyio.run(main)
"

Expected output:

Tool: confirm_delete (task_support=required)
Tool: write_haiku (task_support=required)

---

15. sse-polling-demo (HTTP)

What it does: Demonstrates SSE polling pattern with periodic stream close and client auto-reconnect.

Terminal 1 — Start the server:

cd examples/servers/sse-polling-demo
uv run mcp-sse-polling-demo --port 3000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3000/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools: {[t.name for t in tools.tools]}')
        result = await client.call_tool('process_batch', {'items': 3, 'checkpoint_every': 10})
        print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output:

Tools: ['process_batch']
Result: Successfully processed 3 items with checkpoints every 10 items

Here are the GIFs for Claude running each of these examples:

Open to see GIFs

1. simple-tool (stdio)

2. simple-prompt (stdio)

3. simple-resource (stdio)

4. simple-pagination (stdio)

5. snippet: basic.py (stdio)

6. snippet: direct_call_tool_result.py (stdio)

7. snippet: lifespan.py (stdio)

8. snippet: structured_output.py (stdio)

9. structured-output-lowlevel (stdio)

10. simple-streamablehttp (HTTP)

11. simple-streamablehttp-stateless (HTTP)

12. everything-server (HTTP)

13. simple-task (HTTP)

14. simple-task-interactive (HTTP)

15. sse-polling-demo (HTTP)