Оффлайн
- Регистрация
- 9 Май 2015
- Сообщения
- 1,605
- Баллы
- 155
Introduction
MCP (Model Context Protocol) is a protocol that connects LLMs with external tools. This article walks through how to use MCP step by step with real code examples. The full source code is available in the following GitHub repository:
Read time: ~10 minutes
1. Core Concepts of MCP
MCP provides three primary capabilities:
- Tool: Execute external functionality via function calls
- Resource: Provide data or information (static or dynamic)
- Prompt: Provide prompt templates for LLMs
MCP servers mainly support two transport modes. This article covers both stdio and streamable-http:
- stdio: Communicate via standard I/O (used within the same process)
- streamable-http: Communicate via HTTP (over the network)
Let’s start by building a server using the simplest stdio transport.
2.1 Basic Server Code
from mcp.server.fastmcp import FastMCP
# Create a FastMCP instance
mcp = FastMCP(name="FastMCP Demo Server")
# Define a tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b
# Define a resource (dynamic)
@mcp.resource("time://{zone}")
def get_time(zone: str) -> str:
"""Return ISO 8601 timestamp."""
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
return now.isoformat() if zone.lower() == "utc" else now.astimezone().isoformat()
# Define a resource (static)
@mcp.resource("info://server")
def get_server_info() -> str:
"""Return server metadata."""
return "FastMCP demo server"
# Define a prompt
@mcp.prompt()
def greet_user(name: str, tone: str = "friendly") -> str:
"""Generate a greeting instruction."""
return f"Craft a {tone} greeting addressed to {name}."
# Run the server
if __name__ == "__main__":
mcp.run(transport="stdio")
2.2 Async Tools and Progress Reporting
For long-running operations, you can report progress:
from typing import Annotated
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
import asyncio
@mcp.tool()
async def countdown(
start: int,
ctx: Annotated[Context[ServerSession, None], "Injected by FastMCP"],
) -> list[int]:
"""Count down from start to zero."""
sequence = []
for step, value in enumerate(range(start, -1, -1), start=1):
await ctx.report_progress(
progress=step,
total=start + 1,
message=f"Counting value {value}",
)
sequence.append(value)
await asyncio.sleep(0.2)
return sequence
3. Build a stdio Client
Client code to call the server:
import asyncio
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
server_path = Path(__file__).with_name("server.py")
server_params = StdioServerParameters(command="python", args=[str(server_path)])
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# List tools
tools = await session.list_tools()
print("Available tools:", [tool.name for tool in tools.tools])
# Call a tool
result = await session.call_tool("add", arguments={"a": 2, "b": 5})
print("add result:", result.content)
# Read a resource
resource = await session.read_resource("time://local")
print("time://local:", resource.contents)
# Get a prompt
prompt = await session.get_prompt("greet_user", arguments={"name": "Alice"})
print("prompt:", prompt.messages)
asyncio.run(main())
To receive progress reports, pass progress_callback to ClientSession.call_tool:
async def on_progress(progress: float, total: float | None, message: str | None) -> None:
print(f"{progress}/{total or 0} - {message or ''}")
result = await session.call_tool(
"countdown",
arguments={"start": 3},
progress_callback=on_progress,
)
4. Build a Server with streamable-http
You can convert the stdio server into an HTTP server with just a few changes:
Change: Specify host and port when initializing FastMCP, and use transport="streamable-http".
from mcp.server.fastmcp import FastMCP
# Change: specify host and port
mcp = FastMCP(
name="FastMCP StreamableHTTP Demo",
host="127.0.0.1", # localhost only; 0.0.0.0 also works
port=8765,
)
# Tool, resource, and prompt definitions are the same
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b
# ...(other definitions are the same)
# Change: use transport "streamable-http"
if __name__ == "__main__":
mcp.run(transport="streamable-http")
5. Build a streamable-http Client
The HTTP client also requires only minimal changes:
Change: Use streamablehttp_client instead of stdio_client and specify a URL.
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
STREAMABLE_HTTP_URL = ""
async def main():
# Change: use streamablehttp_client
async with streamablehttp_client(STREAMABLE_HTTP_URL) as (read, write, get_session_id):
async with ClientSession(read, write) as session:
await session.initialize()
# Retrieve the session ID (HTTP-specific)
if (session_id := get_session_id()) is not None:
print("Session ID:", session_id)
# Usage is the same as stdio from here on
tools = await session.list_tools()
print("Available tools:", [tool.name for tool in tools.tools])
result = await session.call_tool("add", arguments={"a": 2, "b": 5})
print("add result:", result.content)
asyncio.run(main())
6. A Client for Multiple Servers
You can handle multiple servers by “keeping and reusing” each ClientSession after establishing connections.
- Open the transport (stdio/HTTP) → create a ClientSession → call initialize → store in a dictionary for lookup by name
# Pattern aligned with the repo implementation (async with)
sessions: dict[str, ClientSession] = {}
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
sessions["stdio"] = session # Keep and reuse if needed
# You can call call_tool as many times as you want here
res = await session.call_tool("add", {"a": 1, "b": 2})
print(res.content)
- Call call_tool on the retained ClientSession
result = await sessions[server].call_tool(tool, args)
- Close everything at the end (centralized lifecycle management for connections and sessions)
This “retain and reuse with centralized lifecycle management” is generalized and organized by MultiServerClient. For implementation details and usage, see multi-server/client.py in the . The following functions/methods are the key points:
- MultiServerClient.__init__: Register servers (stdio/http)
- connect(): Establish connections to registered servers
- _ensure_session(name): Check for a session and connect if missing
- _connect(name): Entry point to create a session (with Future management)
- _session_task(name, future): Actual connection and lifecycle (async with stdio_client/streamablehttp_client → ClientSession → initialize → wait → close)
- session(name): Context manager to retrieve a retained ClientSession
- list_tools(): Enumerate tools from all servers
- call_tool(server, tool, arguments): Delegate execution to the specified server
- close(): Safely terminate all sessions (shutdown events and awaiting completion)
7.1 Which schema to convert to (Responses vs. Chat Completions)
- Responses API (e.g., client.responses.create) expects the following format:
{
"type": "function",
"name": "server__tool",
"description": "...",
"parameters": { "type": "object", "properties": {"a": {"type": "integer"}}, "required": ["a"], "additionalProperties": false },
"strict": true
}
- Chat Completions API (e.g., client.chat.completions.create) expects the following format:
{
"type": "function",
"function": {
"name": "server__tool",
"description": "...",
"parameters": { "type": "object", "properties": {"a": {"type": "integer"}}, "required": ["a"] }
},
"strict": true
}
7.2 How to obtain the necessary information (inspect / docstring / MCP meta)
Generate schemas using information obtained from the MCP server via list_tools().
- Use description and inputSchema
- In this repository, MultiServerClient.get_tool_details() in for-llm/mcp_client.py aggregates list_tools() results across all servers and provides { "description", "inputSchema" }
You can also generate schemas directly from Python functions.
- Compose JSON Schema using inspect, type hints, and docstrings
- The repository’s for-llm/tool_schemas.py implements:
- function_to_responses_tool(func)
- function_to_chat_tool(func)
- build_responses_toolkit(*functions)
- build_chat_toolkit(*functions)
See for-llm/run_llm.py in the for the actual implementation.
Schema conversion utilities: for-llm/tool_schemas.py
- MCP → Responses: mcp_tool_to_responses_schema(tool_name, description, input_schema, strict=True)
- MCP → Chat: mcp_tool_to_chat_schema(tool_name, description, input_schema, strict=True)
- Python function → Responses: function_to_responses_tool(func) / bundle: build_responses_toolkit(...)
- Python function → Chat: function_to_chat_tool(func) / bundle: build_chat_toolkit(...)
Tool metadata retrieval: for-llm/mcp_client.py
- MultiServerClient.get_tool_details() → { combined_name: {"server_name", "tool_name", "description", "inputSchema"} }
- MCP provides three capabilities: tools, resources, and prompts
- stdio communicates within the same process; streamable-http communicates over the network
- Server and client code are very simple
- You can centrally manage multiple servers
- By combining schema conversion with tools, you can dynamically use MCP from generative AI
See the for the full code.