Model Context Protocol (MCP) Architecture
模型上下文协议(MCP)建立在灵活、可扩展的架构之上,该架构使LLM应用程序和集成之间能够实现无缝通信。本文件涵盖了核心架构组件和概念。
Core Architecture
MCP 遵循客户端-服务器架构
Client 1: 1 Server Connect

CleanShot 2025-03-06 at 10.02.18@2x.png
Core Components
Protocol Layer
class Session(BaseSession[RequestT, NotificationT, ResultT]):
async def send_request(
self,
request: RequestT,
result_type: type[Result]
) -> Result:
"""
Send request and wait for response. Raises McpError if response contains error.
"""
# Request handling implementation
async def send_notification(
self,
notification: NotificationT
) -> None:
"""Send one-way notification that doesn't expect response."""
# Notification handling implementation
async def _received_request(
self,
responder: RequestResponder[ReceiveRequestT, ResultT]
) -> None:
"""Handle incoming request from other side."""
# Request handling implementation
async def _received_notification(
self,
notification: ReceiveNotificationT
) -> None:
"""Handle incoming notification from other side."""
# Notification handling implementation
Transport layer
传输层处理客户端和服务器之间的实际通信。MCP 支持多种传输机制:
- Stdio transport 标准 I/O 传输
- 使用标准输入/输出进行通信
- 非常适合本地进程
- HTTP with SSE transport
- 使用服务器发送事件进行服务器到客户端的消息
- HTTP POST 用于客户端到服务器的消息
所有传输都使用 JSON-RPC 2.0 进行消息交换。请参阅规范以获取有关模型上下文协议消息格式的详细信息。
Message types
- Requests
- Response
- Errors
- Notifycations 单向消息,不需要响应
Connection Lifecycle
Initialization 初始化

CleanShot 2025-03-06 at 10.09.25@2x.png
- 客户端发送带有协议版本和功能的
initialize
请求 - 服务器响应其协议版本和功能
- 客户端发送
initialized
通知作为确认 - 普通消息交换开始
Message exchange
初始化后,支持以下模式
- Request-Response: 客户端或服务器发送请求,另一方响应
- Notifications: 任一方发送单向消息
Termination
任何一方都可以终止连接:
- 通过
close()
关闭 - 运输断开
- 错误条件
Error Handling
enum ErrorCode {
// Standard JSON-RPC error codes
ParseError = -32700,
InvalidRequest = -32600,
MethodNotFound = -32601,
InvalidParams = -32602,
InternalError = -32603
}
SDKs 和应用程序可以定义自己的错误代码,其值大于-32000。
Errors通过以下方式传播:
- 错误响应请求
- 传输上的错误事件
- 协议级错误处理器
Example
import asyncio
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server
app = Server("example-server")
@app.list_resources()
async def list_resources() -> list[types.Resource]:
return [
types.Resource(
uri="example://resource",
name="Example Resource"
)
]
async def main():
async with stdio_server() as streams:
await app.run(
streams[0],
streams[1],
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main)
Resources
Resources是模型上下文协议(MCP)中的核心基本元素,允许服务器暴露客户端可以读取并用作LLM交互上下文的数据和内容。
Resources代表 MCP 服务器希望向客户端提供任何类型的数据。这可以包括:
- File contents 文件内容
- Database records 数据库记录
- API responses API 响应
- Live system data 实时系统数据
- Screenshots and images 截图和图像
- Log files 日志文件
- And more
Resource URIs
[protocol]://[host]/[path]
例如:
file:///home/user/documents/report.pdf
postgres://database/customers/schema
screen://localhost/display1
Client discover Resource
Direct resources
{
uri: string; // Unique identifier for the resource
name: string; // Human-readable name
description?: string; // Optional description
mimeType?: string; // Optional MIME type
}
Resource templates
{
uriTemplate: string; // URI template following RFC 6570
name: string; // Human-readable name for this type
description?: string; // Optional description
mimeType?: string; // Optional MIME type for all matching
}
Reading Resources
Response
读取资源时,客户端使用资源 URI 发起一个 resources/read
请求。
服务器响应一个资源内容的列表:
{
contents: [
{
uri: string; // The URI of the resource
mimeType?: string; // Optional MIME type
// One of:
text?: string; // For text resources
blob?: string; // For binary resources (base64 encoded)
}
]
}
Resource updates
- List changes
服务器可以通过notifications/resources/list_changed
通知向客户端通知其可用资源列表的更改。 - Content changes
客户可以订阅特定资源的更新:- 客户端发送
resources/subscribe
与资源 URI - 服务器在资源发生变化时发送
notifications/resources/updated
- 客户端可以使用
resources/read
获取最新内容 - 客户端可以使用
resources/unsubscribe
进行退订
- 客户端发送
Example
app = Server("example-server")
@app.list_resources()
async def list_resources() -> list[types.Resource]:
return [
types.Resource(
uri="file:///logs/app.log",
name="Application Logs",
mimeType="text/plain"
)
]
@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
if str(uri) == "file:///logs/app.log":
log_contents = await read_log_file()
return log_contents
raise ValueError("Resource not found")
# Start server
async with stdio_server() as streams:
await app.run(
streams[0],
streams[1],
app.create_initialization_options()
)
Prompts
Prompts使服务器能够定义可重用的提示模板和工作流程,客户端可以轻松地呈现给用户和LLMs。它们提供了一种强大的方式来标准化和共享常见的LLM交互。
MCP 中的提示是预定义的模板,可以:
- Accept dynamic arguments 接受动态参数
- Include context from resources 包含资源上下文
- Chain multiple interactions 链式多个交互
- Guide specific workflows 指南特定工作流程
- Surface as UI elements (like slash commands) 表面为 UI 元素(如斜杠命令)
Structure
{
name: string; // Unique identifier for the prompt
description?: string; // Human-readable description
arguments?: [ // Optional list of arguments
{
name: string; // Argument identifier
description?: string; // Argument description
required?: boolean; // Whether argument is required
}
]
}
Example
from mcp.server import Server
import mcp.types as types
# Define available prompts
PROMPTS = {
"git-commit": types.Prompt(
name="git-commit",
description="Generate a Git commit message",
arguments=[
types.PromptArgument(
name="changes",
description="Git diff or description of changes",
required=True
)
],
),
"explain-code": types.Prompt(
name="explain-code",
description="Explain how code works",
arguments=[
types.PromptArgument(
name="code",
description="Code to explain",
required=True
),
types.PromptArgument(
name="language",
description="Programming language",
required=False
)
],
)
}
# Initialize server
app = Server("example-prompts-server")
@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
return list(PROMPTS.values())
@app.get_prompt()
async def get_prompt(
name: str, arguments: dict[str, str] | None = None
) -> types.GetPromptResult:
if name not in PROMPTS:
raise ValueError(f"Prompt not found: {name}")
if name == "git-commit":
changes = arguments.get("changes") if arguments else ""
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text=f"Generate a concise but descriptive commit message "
f"for these changes:\n\n{changes}"
)
)
]
)
if name == "explain-code":
code = arguments.get("code") if arguments else ""
language = arguments.get("language", "Unknown") if arguments else "Unknown"
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text=f"Explain how this {language} code works:\n\n{code}"
)
)
]
)
raise ValueError("Prompt implementation not found")
Tools
Tools是模型上下文协议(MCP)中的一个强大原始功能,它使服务器能够向客户端暴露可执行功能。通过工具,LLMs可以与外部系统交互,执行计算并在现实世界中采取行动。
MCP 中的工具允许服务器暴露可由客户端调用并由LLMs执行操作的执行函数。工具的关键方面包括:
- Discovery: 客户端可以通过
tools/list
端点列出可用的工具 - Invocation: 使用
tools/call
端点调用工具,服务器执行请求的操作并返回结果 - Flexibility: 工具可以从简单的计算到复杂的 API 交互
Structure
{
name: string; // Unique identifier for the tool
description?: string; // Human-readable description
inputSchema: { // JSON Schema for the tool's parameters
type: "object",
properties: { ... } // Tool-specific parameters
}
}
Example
app = Server("example-server")
@app.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="calculate_sum",
description="Add two numbers together",
inputSchema={
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
)
]
@app.call_tool()
async def call_tool(
name: str,
arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
if name == "calculate_sum":
a = arguments["a"]
b = arguments["b"]
result = a + b
return [types.TextContent(type="text", text=str(result))]
raise ValueError(f"Tool not found: {name}")
Roots
Roots是 MCP 中的一个概念,定义了服务器可以操作的范围。它们提供了一种方式,让客户端向服务器告知相关资源及其位置。
Roots是一个客户端建议服务器应关注的 URI。当客户端连接到服务器时,它会声明服务器应与哪些Roots一起工作。虽然主要用于文件系统路径,但根可以是任何有效的 URI,包括 HTTP URL。
file:///home/user/projects/myapp
https://api.example.com/v1
根节点具有几个重要功能:
- Guidance: 它们向服务器通知相关资源和位置
- Clarity: 根资源明确哪些资源属于您的工作空间
- Organization: 多个根让您可以同时处理不同的资源
Example
{
"roots": [
{
"uri": "file:///home/user/projects/frontend",
"name": "Frontend Repository"
},
{
"uri": "https://api.example.com/v1",
"name": "API Endpoint"
}
]
}