Model Context Protocol (MCP) Architecture

模型上下文协议（MCP）建立在灵活、可扩展的架构之上，该架构使LLM应用程序和集成之间能够实现无缝通信。本文件涵盖了核心架构组件和概念。

Core Architecture

MCP 遵循客户端-服务器架构

Client 1: 1 Server Connect

Core Components

Protocol Layer

PYTHON

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 初始化

客户端发送带有协议版本和功能的 initialize 请求
服务器响应其协议版本和功能
客户端发送 initialized 通知作为确认
普通消息交换开始

Message exchange

初始化后，支持以下模式

Request-Response: 客户端或服务器发送请求，另一方响应
Notifications: 任一方发送单向消息

Termination

任何一方都可以终止连接：

通过 close()关闭
运输断开
错误条件

Error Handling

PYTHON

enum ErrorCode {
  // Standard JSON-RPC error codes
  ParseError = -32700,
  InvalidRequest = -32600,
  MethodNotFound = -32601,
  InvalidParams = -32602,
  InternalError = -32603
}

SDKs 和应用程序可以定义自己的错误代码，其值大于-32000。

Errors通过以下方式传播：

错误响应请求
传输上的错误事件
协议级错误处理器

Example

PYTHON

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

URI

[protocol]://[host]/[path]

例如：

file:///home/user/documents/report.pdf
postgres://database/customers/schema
screen://localhost/display1

Client discover Resource

Direct resources

JSON

{
  uri: string;           // Unique identifier for the resource
  name: string;          // Human-readable name
  description?: string;  // Optional description
  mimeType?: string;     // Optional MIME type
}

Resource templates

JSON

{
  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 请求。

服务器响应一个资源内容的列表：

JSON

{
  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
客户可以订阅特定资源的更新：
1. 客户端发送 resources/subscribe 与资源 URI
2. 服务器在资源发生变化时发送 notifications/resources/updated
3. 客户端可以使用 resources/read 获取最新内容
4. 客户端可以使用 resources/unsubscribe 进行退订

Example

PYTHON

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

JSON

{
  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

PYTHON

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

JSON

{
  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

PYTHON

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。

TXT

file:///home/user/projects/myapp
https://api.example.com/v1

根节点具有几个重要功能：

Guidance: 它们向服务器通知相关资源和位置
Clarity: 根资源明确哪些资源属于您的工作空间
Organization: 多个根让您可以同时处理不同的资源

Example

JSON

{
  "roots": [
    {
      "uri": "file:///home/user/projects/frontend",
      "name": "Frontend Repository"
    },
    {
      "uri": "https://api.example.com/v1",
      "name": "API Endpoint"
    }
  ]
}