1. 引言：从认知架构说起

当我们谈论大语言模型（LLM）的“认知架构”时，本质上是在回答一个问题：模型如何组织、检索和利用信息，从而完成复杂的推理与决策？早期的做法是把所有上下文一股脑塞进提示词，但随着应用复杂度提升，这种“填鸭式”方法很快触及上下文窗口的天花板。

模型上下文协议（Model Context Protocol，简称 MCP）正是为解决这一瓶颈而生的开放标准。它把“模型需要什么数据”和“数据从哪里来”解耦，让模型通过统一的协议接口，按需、安全地访问外部工具、数据源和业务系统。可以说，MCP 是连接模型认知能力与外部世界的关键桥梁。

2. MCP 核心架构

MCP 采用客户端-服务器（Client-Server）架构，包含三个核心角色：

• MCP Host：运行 LLM 应用的主进程，例如 Claude Desktop、IDE 插件或自定义应用，负责与用户交互并调度模型。
• MCP Client：嵌入在 Host 内部，负责与 MCP Server 建立连接、发送请求并接收响应。
• MCP Server：轻量级服务进程，暴露标准化的工具（Tools）、资源（Resources）和提示词（Prompts），供模型调用。

三者协作流程如下：

1. Host 启动时，MCP Client 根据配置连接一个或多个 MCP Server。
2. 连接建立后，Client 向 Server 获取能力清单（工具、资源、提示词列表）。
3. 模型在推理过程中，通过 Client 向 Server 发起工具调用或资源读取请求。
4. Server 执行实际操作（查询数据库、调用 API、读写文件等），将结果返回给模型。

这种架构带来的核心收益是：模型不再需要预装所有知识，而是像人一样“按需查阅”，从而大幅降低上下文占用，提升回答的准确性和时效性。

3. 传输层与消息格式

MCP 目前支持两种传输方式：

• stdio：通过标准输入输出进行通信，适用于本地进程，例如本地文件系统工具。
• Streamable HTTP：基于 HTTP 的流式传输，适用于远程服务，例如云端数据库或第三方 API。

所有消息均采用 JSON-RPC 2.0 格式封装。一个典型的请求消息结构如下：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "sql": "SELECT * FROM users WHERE id = 42"
    }
  }
}

响应消息则包含执行结果或错误信息：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "{\"id\":42,\"name\":\"Alice\",\"email\":\"alice@example.com\"}"
      }
    ]
  }
}

4. 核心原语：工具、资源与提示词

MCP 定义了三种核心原语，分别对应模型认知的不同需求：

4.1 工具（Tools）

工具是模型可以主动调用的函数，用于执行操作。例如查询天气、发送邮件、执行代码等。工具通常有明确的输入参数和输出格式，模型根据用户意图自主决定是否调用。

4.2 资源（Resources）

资源是只读的数据源，例如文件内容、数据库记录、API 响应等。模型可以读取资源来获取上下文，但不能修改它们。资源通过 URI 定位，例如 file:///etc/config.json 或 db://users/42。

4.3 提示词（Prompts）

提示词是可复用的提示模板，封装了特定任务的指令和上下文。例如“代码审查提示词”可以包含审查标准、输出格式等，模型加载后即可按模板执行任务。

5. 环境准备与项目初始化

在开始代码实战之前，我们需要准备开发环境。以下以 Python 为例，使用官方 SDK 构建一个完整的 MCP 服务。

首先，创建项目目录并初始化虚拟环境：

mkdir mcp-demo
cd mcp-demo
python -m venv venv
source venv/bin/activate  # Windows 下使用 venv\Scripts\activate

安装 MCP SDK 和依赖：

pip install mcp httpx

6. 实战：构建一个天气查询 MCP Server

下面我们构建一个完整的天气查询服务，演示如何定义工具、处理请求并返回结构化结果。

6.1 定义工具 Schema

首先创建 weather_server.py，定义天气查询工具：

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
import json
app = Server("weather-server")
@app.list_tools()
async def list_tools():
return [
Tool(
name="get_weather",
description="查询指定城市的实时天气",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，例如北京、上海"
}
},
"required": ["city"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "get_weather":
city = arguments.get("city", "")
# 这里使用模拟数据，实际可替换为真实天气 API
weather_data = {
"city": city,
"temperature": 26,
"condition": "晴",
"humidity": 45
}
return [
TextContent(
type="text",
text=json.dumps(weather_data, ensure_ascii=False)
)
]
raise ValueError(f"未知工具: {name}")
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if name == "main":
import asyncio
asyncio.run(main())

6.2 运行 Server

在终端启动服务：

python weather_server.py

服务启动后，会通过 stdio 等待 MCP Client 的连接和调用。

7. 实战：构建 MCP Client 并调用工具

接下来，我们编写一个 MCP Client，连接上述 Server 并调用天气查询工具。

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 配置 Server 启动参数
server_params = StdioServerParameters(
command="python",
args=["weather_server.py"],
env=None
)
async with stdio_client(server_params) as (read_stream, write_stream):
    async with ClientSession(read_stream, write_stream) as session:
        # 初始化连接
        await session.initialize()
    # 获取工具列表
    tools = await session.list_tools()
    print("可用工具:", [t.name for t in tools.tools])
# 调用天气查询工具
result = await session.call_tool(
    "get_weather",
    {"city": "北京"}
)
解析并打印结果
for content in result.content:
if content.type == "text":
print("查询结果:", content.text)
if name == "main":
asyncio.run(main())

运行客户端：

python weather_client.py

预期输出：

可用工具: ['get_weather']
查询结果: {"city": "北京", "temperature": 26, "condition": "晴", "humidity": 45}

8. 实战：集成 LLM 实现认知闭环

上面的例子展示了 Client 和 Server 的通信，但还没有真正接入 LLM。下面我们演示如何让 LLM 自主决定调用工具，形成完整的认知闭环。

我们使用 OpenAI 兼容接口，让模型根据用户问题决定是否调用天气工具：

import asyncio
import json
from openai import AsyncOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
client = AsyncOpenAI(
base_url="https://api.openai.com/v1",
api_key="YOUR_API_KEY"
)
async def chat_with_tools(user_query: str):
server_params = StdioServerParameters(
command="python",
args=["weather_server.py"],
env=None
)
async with stdio_client(server_params) as (read_stream, write_stream):
    async with ClientSession(read_stream, write_stream) as session:
        await session.initialize()
    # 获取工具定义并转换为 OpenAI 格式
    tools_result = await session.list_tools()
    openai_tools = []
    for tool in tools_result.tools:
        openai_tools.append({
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            }
        })
# 第一轮：让模型决定是否调用工具
messages = [{"role": "user", "content": user_query}]
response = await client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=openai_tools,
    tool_choice="auto"
)
assistant_msg = response.choices[0].message
如果模型决定调用工具
if assistant_msg.tool_calls:
for tool_call in assistant_msg.tool_calls:
fn_name = tool_call.function.name
fn_args = json.loads(tool_call.function.arguments)
    # 通过 MCP 调用工具
    mcp_result = await session.call_tool(fn_name, fn_args)
    tool_output = ""
    for content in mcp_result.content:
        if content.type == "text":
            tool_output += content.text

    # 将工具结果返回给模型
    messages.append(assistant_msg)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": tool_output
    })

# 第二轮：让模型基于工具结果生成最终回答
final_response = await client.chat.completions.create(
    model="gpt-4o",
    messages=messages
)
return final_response.choices[0].message.content
return assistant_msg.content
async def main():
result = await chat_with_tools("北京今天天气怎么样？")
print("AI 回答:", result)
if name == "main":
asyncio.run(main())

运行后，模型会先调用 get_weather 工具获取北京天气，再基于返回数据生成自然语言回答，例如：

AI 回答: 北京今天天气晴朗，气温 26 摄氏度，湿度 45%，体感舒适。

9. 进阶：资源与提示词实战

除了工具，MCP 还支持资源和提示词。下面演示如何在 Server 中暴露一个只读资源和一个提示词模板。

from mcp.types import Resource, Prompt, PromptArgument
在 weather_server.py 中追加以下代码
@app.list_resources()
async def list_resources():
return [
Resource(
uri="weather://default/city-list",
name="支持的城市列表",
description="当前支持查询天气的城市清单",
mimeType="application/json"
)
]
@app.read_resource()
async def read_resource(uri: str):
if uri == "weather://default/city-list":
data = json.dumps(["北京", "上海", "广州", "深圳"], ensure_ascii=False)
return TextContent(type="text", text=data)
raise ValueError(f"未知资源: {uri}")
@app.list_prompts()
async def list_prompts():
return [
Prompt(
name="weather_report",
description="生成指定城市的天气播报文案",
arguments=[
PromptArgument(
name="city",
description="城市名称",
required=True
)
]
)
]
@app.get_prompt()
async def get_prompt(name: str, arguments: dict | None = None):
if name == "weather_report":
city = arguments.get("city", "北京") if arguments else "北京"
prompt_text = f"请根据以下天气数据，生成一段适合广播的天气播报文案，城市：{city}。"
return Prompt(
name="weather_report",
description="生成天气播报文案",
arguments=[
PromptArgument(name="city", description="城市名称", required=True)
]
)
raise ValueError(f"未知提示词: {name}")

客户端可以通过 session.list_resources() 和 session.list_prompts() 获取这些能力，并在需要时读取资源或加载提示词。

10. 安全与最佳实践

在生产环境中使用 MCP，需要关注以下安全与工程实践：

• 权限最小化：每个 MCP Server 只暴露必要的工具和资源，避免过度授权。
• 输入校验：在 Server 端对所有工具参数进行严格校验，防止注入攻击。
• 超时与重试：为工具调用设置合理的超时时间，并对失败调用实现重试机制。
• 日志与监控：记录所有工具调用日志，便于审计和排查问题。
• 敏感信息保护：不要在工具参数或返回结果中泄露 API Key、密码等敏感信息。

11. 总结与展望

MCP 为 LLM 应用提供了一套标准化的认知扩展框架，让模型能够按需访问外部工具和数据，从而突破上下文窗口的限制，实现更复杂、更可靠的智能应用。通过本文的代码实战，你已经掌握了构建 MCP Server、Client 以及集成 LLM 的完整流程。

未来，随着 MCP 生态的成熟，我们可以期待更多开箱即用的 Server 实现，以及更丰富的工具、资源和提示词共享机制。认知架构的边界，将不再受限于模型本身的参数和上下文，而是由整个生态的连接广度所决定。