从零开始的 LLM Agent编程生活--MCP 篇

[LuYanFCP]

1. 什么是MCP(Model Context Protocol)

MCP（Model Context Protocol，模型上下文协议）是由Anthropic公司于2024年推出并开源的一种通信协议，旨在解决大型语言模型（LLM）与外部数据源及工具之间的连接问题。定义了 Model 与外部接口/数据/Prompt 通信之间的协议。 工具/资源提供方只需要实现 MCP 协议就可以和实现 MCP 客户端的 LLM APP 链接，LLM APP在运行过程中自动根据协议中返回工具列表/Prompt/资源列表，通过JsonRpc 从 MCP 服务器中获取。

一个简单的例子：接入高德地图，让模型通过高德地图的 API 查询相关天气/路径/地图周围相关信息的能力。

1. 没有 MCP：需要自己实现对高德 OpenAPI 的调用 Tools，自己写 Prompt 组织对 LLM 的请求。
2. 有 MCP 后：直接使用 MCP Client 写上高德 MCP Service 的 Endpoint 和 Key，LLM 会在运行期间主动通过 MCP 查询高德相关资源，并使用高德已经组织好的 Prompt 反馈给 LLM。

2. MCP定义了什么：

MCP 定义的原语：

1. Tools：FunctionCall
2. Resouce：资源
3. Prompts：提供结构化模板
4. Sampling：允许服务器请求客户端调用 LLM

2.1 工具：

经常有人拿 FunctionCall 和 MCP 做对比，甚至发出既生FunctionCall 又生 MCP 的感叹，我个人认为 FunctionCall和 MCP 并不冲突，FunctionCall 其实是 MCP 的一个子集，MCP 也支持 FunctionCall，只不过 MCP 还支持 Resource/Prompt 等定义，并在协议层对其获取/调用/更新做了明确 Protocol 约束。

MCP从协议层定义了工具的获取，调用协议：

1. tools/list 获取当前 MCP 所有的工具列表，主要是元数据，工具的描述和需要的参数，以及输出结果的 schema。
2. tools/call 执行调用工具的动作，并获取结果。
3. notifications/tools/list_changed 长链接中通过 Push 的方式更新 Client 端缓存的 Tools 信息。MCP Server -> Client

在 Agent 中使用他的方式一般和传统的 FunctionCall 方式一致，链接 MCP 之后获取到所有 Tools 的 Meta 之后直接将其渲染到 SystemPrompt 上即可。

2.2 资源

Resource 在 MCP 中是一种应用程序控制的原语，允许服务器向客户端公开可被读取的数据和内容，这些内容可用作 LLM（大型语言模型）交互的上下文。资源类似于 Restful 接口中对资源的定义，其可以是文件/数据库记录/API 响应/日志文件。

MCP 要求每个资源实体都要有一个唯一的 URI，格式是标准的 URL 格式protocol://host/path，处理有 URI，例如如果要讲postgres 中某个表以资源的形式对外开放，他的 URI 为postgres://<host>:5432/<schema>/<database>/<table>。

在 MCP中资源的元数据定义为:

export interface Resource {
  uri: string;
  name: string;
  description?: string;
  mimeType?: string;
  annotations?: Annotations;
  size?: number;
}

资源其实是整体的获取的，这也是他与 FunctionCall 最不同的地方，比如说在数据库这个场景，如果我实现一个 QueryTools 其实也可以实现类似于资源的效果，但是资源更强调一次性将资源所有信息返回，而 Tools 强调做了一些动作获得的结果，动作可能是 read 也可能是write。

2.3 Prompt

不同的 MCP Server 提供的特有的 Prompt 模板，只要是针对当前 MCP 提供的功能配套的 Prompt，一般会可以 Prompt 快速的让 LLM 更好的调用 MCP Server 中工具的能力。

例如一个重构代码的 Agent：

1. Resource：其提供的 MCP Server 的资源，一般都是本地的代码文件，以及代码规范的标准文件。
2. Tools：一般是本地的 Lint工具对特定文件的检查结果。比如说analyze_code_complexity/check_code_standards/check_code_type
3. Prompt: 一般提供独特的代码规范 Prompt，以及大模型应该如何使用工具，例如他可以提供模型，在重构的时候依次调用check_code_standards->check_code_standards->check_code_type

2.4 Sampling

Sampling 是 MCP 中允许服务器向客户端请求 LLM 完成的功能。这与传统的交互模式形成鲜明对比：通常是客户端向服务器请求数据或功能，但在 Sampling 中，服务器可以主动请求客户端调用 LLM 来生成文本或进行推理。

简单来说，Sampling 允许 MCP 服务器"反向"使用客户端连接的语言模型，从而实现更复杂的 AI 代理行为，同时保持安全性和隐私控制.

Sampling 的工作流程遵循以下步骤 :

1. 服务器发起请求：服务器发送 sampling/createMessage 请求给客户端
2. 客户端审核：客户端检查请求并可以修改它
3. LLM 调用：客户端调用 LLM 并获取完成结果
4. 客户端审核结果：客户端检查 LLM 生成的内容
5. 返回结果：客户端将结果返回给服务器。

服务器在请求 Sampling 时可以提供多种参数来微调 LLM 的行为 :

• temperature：控制随机性（0.0 到 1.0）
• maxTokens：最大生成的标记数
• stopSequences：停止生成的序列数组
• metadata：额外的特定于提供商的参数

服务器还可以通过 modelPreferences 对象来指定模型选择偏好，以及通过 systemPrompt 字段请求特定的系统提示，但客户端最终决定使用什么模型以及是否采纳系统提示 。

Sampling 特别适用于需要"代理行为"(agentic behavior)的场景，即服务器需要 LLM 的帮助来完成任务。典型用例包括：

1. Git 服务工具：请求 LLM 基于代码差异编写提交消息。
2. 数据分析服务：请求 LLM 解释数据分析结果并提供洞见。
3. 内容生成：为特定领域工具生成文本内容，如邮件草稿或文档摘要。
4. 复杂决策：基于服务器提供的特定领域数据请求 LLM 做出决策推荐。

一个 Case

sequenceDiagram
    autonumber
    actor User as 开发者
    participant Client as Git MCP 客户端
    participant Server as Git MCP 服务器
    participant LLM as 语言模型
    
    Note over User: 修改代码并暂存更改
    User->>Client: 请求生成提交消息
    Client->>Server: 调用 generate_commit_message 工具
    
    Note over Server: 收集上下文信息
    Server->>Server: 获取已暂存更改 (git diff --cached)
    Server->>Server: 获取修改文件列表 (git status)
    Server->>Server: 获取最近提交历史 (git log)
    
    rect rgb(240, 248, 255)
        Note over Server, Client: Sampling 流程开始
        Server->>Client: sampling/createMessage 请求
        Note right of Server: 包含代码差异、文件列表<br/>提交历史等上下文
        
        Client->>User: 显示采样请求并征求确认
        User->>Client: 确认允许使用 LLM
        
        Client->>LLM: 调用 LLM API
        LLM->>Client: 返回生成的提交消息
        
        Client->>User: 显示生成的提交消息
        User->>Client: 接受/编辑/拒绝消息
        
        Client->>Server: 返回采样结果 (最终提交消息)
        Note over Server, Client: Sampling 流程结束
    end
    
    Server->>Client: 返回生成的提交消息
    Client->>User: 显示生成的消息并询问是否提交
    
    alt 用户确认提交
        User->>Client: 确认提交
        Client->>Server: 调用 commit_changes 工具
        Server->>Server: 执行 git commit 命令
        Server->>Client: 返回提交结果
        Client->>User: 显示提交成功信息
    else 用户取消
        User->>Client: 取消提交
        Client->>User: 显示操作已取消
    end

Loading

3. 一个现代的 Mcp 示例

3.1 Sequential Thinking MCP + 使用 MCP Function 做状态干预

Sequential Thinking MCP 是一个标准的 MCP 示例，他通过 function 的方式引导LLM 去分步思考并得出结论。他的 Tool Prompt 模板为：

A detailed tool for dynamic and reflective problem-solving through thoughts.
This tool helps analyze problems through a flexible thinking process that can adapt and evolve.
Each thought can build on, question, or revise previous insights as understanding deepens.

When to use this tool:
- Breaking down complex problems into steps
- Planning and design with room for revision
- Analysis that might need course correction
- Problems where the full scope might not be clear initially
- Problems that require a multi-step solution
- Tasks that need to maintain context over multiple steps
- Situations where irrelevant information needs to be filtered out

Key features:
- You can adjust total_thoughts up or down as you progress
- You can question or revise previous thoughts
- You can add more thoughts even after reaching what seemed like the end
- You can express uncertainty and explore alternative approaches
- Not every thought needs to build linearly - you can branch or backtrack
- Generates a solution hypothesis
- Verifies the hypothesis based on the Chain of Thought steps
- Repeats the process until satisfied
- Provides a correct answer

Parameters explained:
- thought: Your current thinking step, which can include:
* Regular analytical steps
* Revisions of previous thoughts
* Questions about previous decisions
* Realizations about needing more analysis
* Changes in approach
* Hypothesis generation
* Hypothesis verification
- next_thought_needed: True if you need more thinking, even if at what seemed like the end
- thought_number: Current number in sequence (can go beyond initial total if needed)
- total_thoughts: Current estimate of thoughts needed (can be adjusted up/down)
- is_revision: A boolean indicating if this thought revises previous thinking
- revises_thought: If is_revision is true, which thought number is being reconsidered
- branch_from_thought: If branching, which thought number is the branching point
- branch_id: Identifier for the current branch (if any)
- needs_more_thoughts: If reaching end but realizing more thoughts needed

You should:
1. Start with an initial estimate of needed thoughts, but be ready to adjust
2. Feel free to question or revise previous thoughts
3. Don't hesitate to add more thoughts if needed, even at the "end"
4. Express uncertainty when present
5. Mark thoughts that revise previous thinking or branch into new paths
6. Ignore information that is irrelevant to the current step
7. Generate a solution hypothesis when appropriate
8. Verify the hypothesis based on the Chain of Thought steps
9. Repeat the process until satisfied with the solution
10. Provide a single, ideally correct answer as the final output
11. Only set next_thought_needed to false when truly done and a satisfactory answer is reached

以下以经典弱智吧问题"不能直接喝，苹果不能直接吃，为什么让不能直接喝的水洗洗苹果就能吃了？"作为提问展示整个引导过程。

sequenceDiagram
    participant User as 用户
    participant LLM as 大语言模型
    participant MCP as fwZGTnwmPNgMz0ViR9Umm8 (MCP)

    User->>LLM: 提问: 水不能直接喝，苹果不能直接吃，<br/>为什么让不能直接喝的水洗洗苹果就能吃了？

    LLM->>LLM: 内部思考: "这个问题涉及到的是对事物属性的理解和逻辑推理...<br/>...让不能直接喝的水用来洗苹果后苹果能吃的原因是因为清洗的目的在于清洁苹果而非改变水的性质。"

    Note over LLM,MCP: 开始调用fwZGTnwmPNgMz0ViR9Umm8工具进行逐步思考

    LLM->>MCP: 调用fwZGTnwmPNgMz0ViR9Umm8工具<br/>{<br/>  "thought": "不能直接喝的水的主要问题是含有不适合直接饮用的物质，<br/>这些物质不会转移到苹果上，清洗后的苹果依然可以安全食用。",<br/>  "nextThoughtNeeded": true,<br/>  "thoughtNumber": 1,<br/>  "totalThoughts": 3,<br/>  "isRevision": false,<br/>  "revisesThought": null,<br/>  "branchFromThought": null,<br/>  "branchId": null,<br/>  "needsMoreThoughts": true<br/>}
    MCP->>LLM: 返回思考步骤状态<br/>{<br/>  "thoughtNumber": 1,<br/>  "totalThoughts": 3,<br/>  "nextThoughtNeeded": true,<br/>  "branches": [],<br/>  "thoughtHistoryLength": 1<br/>}

    LLM->>MCP: 调用fwZGTnwmPNgMz0ViR9Umm8工具<br/>{<br/>  "thought": "不能直接喝的水的主要问题是含有不适合直接饮用的物质，<br/>这些物质不会转移到苹果上，清洗后的苹果依然可以安全食用。",<br/>  "nextThoughtNeeded": true,<br/>  "thoughtNumber": 1, <br/>  "totalThoughts": 3,<br/>  "isRevision": false,<br/>  "revisesThought": null,<br/>  "branchFromThought": null,<br/>  "branchId": null,<br/>  "needsMoreThoughts": true<br/>}
    MCP->>LLM: 返回思考步骤状态<br/>{<br/>  "thoughtNumber": 2,<br/>  "totalThoughts": 3,<br/>  "nextThoughtNeeded": true,<br/>  "branches": [],<br/>  "thoughtHistoryLength": 2<br/>}

    LLM->>MCP: 调用fwZGTnwmPNgMz0ViR9Umm8工具<br/>{<br/>  "thought": "不能直接喝的水的主要问题是含有不适合直接饮用的物质，<br/>这些物质不会转移到苹果上，清洗后的苹果依然可以安全食用。",<br/>  "nextThoughtNeeded": true,<br/>  "thoughtNumber": 1,<br/>  "totalThoughts": 3,<br/>  "isRevision": false,<br/>  "revisesThought": null,<br/>  "branchFromThought": null,<br/>  "branchId": null,<br/>  "needsMoreThoughts": true<br/>}
    MCP->>LLM: 返回思考步骤状态<br/>{<br/>  "thoughtNumber": 3,<br/>  "totalThoughts": 3,<br/>  "nextThoughtNeeded": false,<br/>  "branches": [],<br/>  "thoughtHistoryLength": 3<br/>}

    Note over LLM,MCP: 思考过程完成

    LLM->>User: 提供最终答案: "通过使用fwZGTnwmPNgMz0ViR9Umm8工具进行逐步思考...<br/>...最终答案是清洗苹果的水没有变成可以直接饮用的水，但苹果表面的污染物被去除，使其可以安全食用。"

Loading

4. Transport 层

Agent 系统通过 JsonRpc 与 McpClient 进行通信：
两种模式：

1. stdio 管道： 协议整体的逻辑来源于 Language Server Protocol，Agent 启动时通过启动子进程的方式拉起 MCP Client，Agent 通过stdio 管道中发送 JsonRpc 消息与 MCP 服务进行通信。
2. HTTP-SSE/Streamable-HTTP 服务：允许通过 HTTP Stream进行远端通信，老协议使用 HTTP-SSE 协议（HTML5），通过与服务器的 HTTP Stream 长链接允许与 MCP Server 进行双向通信。25 年 4 月开始支持 Streamable-HTTP（https://github.com/modelcontextprotocol/modelcontextprotocol/pull/206）， 废弃之前的 HTTP-SSE协议。Streamable-HTTP 对 FC 这种计算形态更加优化。

为什么要使用 HTTP-SSE 这么奇怪的 事件 Endpoint 和消息 Endpoint 分离的方式：

The separation of the session establishment and messaging endpoints is intended to simplify Cross-Origin Resource Sharing (CORS). By > providing a 'simple' HTTP POST endpoint for message exchange, CORS preflight requests can be avoided

1. MCP 主要使用场景在浏览器上。如果不分离 Endpoint 的话，HTTP 头上会携带 SessionID 信息，不满足浏览器的 Simple Request 的要求，需要做 Cros 预检查【OPTIONS】
2. 通过 Endpoint 分离的方式，将所有的请求都能变成 “简单请求”，不会触发 OPTIONS 检查
3. 为什么后续 Stream-HTTP 又放弃这种方式：
   a. 认为现代Web开发中CORS预检的性能影响已不是主要问题
   b. 端点分离带来的实现复杂性超过了避免预检的收益
   c. 提供了更清晰的会话管理机制(通过Mcp-Session-Id头部)
   Cloudflare引入：[RFC] Replace HTTP+SSE with new "Streamable HTTP" transport modelcontextprotocol/modelcontextprotocol#206

HTTP-SSE Client Python实现：

1. 使用线程同步编程的实现会需要使用单独一个线程于 Msg Endpoint 建立 HTTP-Stream 长链接用于获取 JsonRpc 返回Event ，并将返回的内容通过回调函数+队列的方式通知给主线程收割事件。

sequenceDiagram
    %% 定义参与者样式
    participant Client as 客户端
    participant MsgEndpoint as 服务器消息端点
    participant SSEEndpoint as 服务器SSE事件端点

    rect rgb(240, 240, 255)
    Note over Client,MsgEndpoint: 获取 Endpoint
    Client->>MsgEndpoint: 发起HTTP POST请求建立连接(Stream=True)
    MsgEndpoint-->>Client: 返回HTTP 200响应, 将 Endpoint 地址返回，URL 中包含唯一 SessionId
    end

    rect rgb(240, 240, 255)
    Note over Client,SSEEndpoint: 初始化，从服务期获取 MCP元数据
    Client->>SSEEndpoint: Rpc Post 请求，{"method": "initialize", "jsonrpc": "2.0", "id": 1}
    SSEEndpoint-->>Client: 返回HTTP 200 
    MsgEndpoint-->>Client: 返回JsonRpc Response  {"jsonrpc": "2.0", "id": 1, "result": ...}
    Client->>SSEEndpoint: NotifyRPC，不需要返回{"method": "notifications/initialized", "jsonrpc": "2.0"}    
    end

    rect rgb(240, 240, 255)
    Note over Client,SSEEndpoint: 获取工具列表
    Client->>SSEEndpoint: Rpc Post 请求，{"method": "tools/list", "jsonrpc": "2.0", "id": 2}
    SSEEndpoint-->>Client: 返回HTTP 200
   MsgEndpoint-->>Client: 返回JsonRpc Response  {"jsonrpc": "2.0", "id": 1, "result": ...}
   end

  rect rgb(240, 240, 255)
    Note over Client,SSEEndpoint: 调用工具
   Client ->> SSEEndpoint: Rpc Post 请求，{"method": "tools/call, "jsonrpc": "2.0", "id": 3, params: {name: "get_weather",arguments: {"location": "Beijing"}}
    SSEEndpoint-->>Client: 返回HTTP 200
   MsgEndpoint-->>Client: 返回JsonRpc Response  {"jsonrpc": "2.0", "id": 3, "result": ...}
end
    
    

Loading

一个简单 python 实现

import os
import queue
import threading
import logging
from typing import Any, Callable, Literal

import json

from urllib.parse import urljoin
from pydantic import BaseModel, ConfigDict
import requests

logger = logging.getLogger(__name__)


class Message(BaseModel):
    event_type: Literal["message"]
    data: str


class JsonRpcHeader(BaseModel):
    method: str
    jsonrpc: str = "2.0"


class JsonRpcRequest(JsonRpcHeader):
    params: dict | None = None
    id: int


class JsonRpcNotify(JsonRpcHeader): ...


type JsonRpcMessage = JsonRpcHeader | JsonRpcNotify


class JSONRPCResponse(BaseModel):
    """A successful (non-error) response to a request."""

    jsonrpc: str
    id: int
    result: dict[str, Any]
    model_config = ConfigDict(extra="allow")


class SSEClient(threading.Thread):
    def __init__(self, url: str) -> None:
        super().__init__()
        self._url = url
        self._queue = queue.Queue()
        self._ready_event = threading.Event()
        self._endpoint = ""
        self._callback = {}

    def run(self) -> None:
        print("run")
        self._event_loop()

    def wait_ready_for_endpoint(self) -> str:
        self._ready_event.wait()
        return self._endpoint

    def register_callback(
        self, message_type: str, callback: Callable[[Message], None]
    ) -> None:
        self._callback[message_type] = callback

    def _event_loop(self) -> None:
        response = requests.get(
            self._url,
            stream=True,
        )

        if response.status_code not in (200, 202):
            raise Exception(f"Failed to connect to server: {response.status_code}")

        event_type = None
        event_data = None

        for line in response.iter_lines(chunk_size=2, decode_unicode=True):
            print(line)
            if line.startswith("event:"):
                event_type = line[6:].strip()
            if line.startswith("data:"):
                event_data = line[5:].strip()

            if event_data is not None and event_type is not None:
                match event_type:
                    case "message":
                        msg = Message(event_type=event_type, data=event_data)
                        if "message" in self._callback:
                            try:
                                self._callback["message"](msg)
                            except Exception as e:
                                logger.error(e)
                        else:
                            self._queue.put(msg)

                    case "endpoint":
                        event_data = event_data.strip()
                        self._endpoint = event_data
                        self._ready_event.set()

                event_type, event_data = None, None


class McpClient:
    def __init__(
        self, sse: SSEClient, endpoint: str, session: requests.Session
    ) -> None:
        self._endpoint = endpoint
        self._sse = sse
        self._sess = session
        self.capabilities = {}
        self._id = 0
        self._callback_record = {}
        self._sse.register_callback("message", self.notify_callback)

    def initialize(self) -> None:
        result: dict = self._send_request(
            method="initialize",
            params={
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {"call": True}, "resources": {"read": True}},
                "clientInfo": dict(name="mcp", version="0.1.0"),
            },
        )
        self.capabilities = result.get("capabilities", {})
        self._send_request(method="notifications/initialized")

    def get_tools(self) -> dict:
        return self._send_request(method="tools/list")
    
    def call_tool(self, tool_name: str, params: dict) -> dict:
        return self._send_request(
            method="tools/call",
            params={
                "name": tool_name,
                "arguments": params,
            },
        )

    def notify_callback(self, message: Message) -> None:
        response = JSONRPCResponse.model_validate(json.loads(message.data))
        if response.id in self._callback_record:
            self._callback_record[response.id](response)

    def _send_request(self, method: str, params: dict | None = None) -> dict | None:

        event = threading.Event()
        result = None
        if "notifications" in method:
            data = JsonRpcNotify(
                method=method,
            )
        else:
            data = JsonRpcRequest(
                method=method,
                params=params,
                id=self._id,
            )

            def get_result(item):
                nonlocal result
                event.set()
                result = item.result

            self._callback_record[self._id] = get_result
            self._id += 1

        print(f"request: url:{self._endpoint} data:{data.json()}")

        response = self._sess.post(
            url=self._endpoint,
            json=data.model_dump(
                mode="json",
                exclude_none=True,
            ),
        )

        if response.status_code not in (200, 202):
            raise Exception(f"Connect Error: {response.text}")

        if "notifications" in method:
            return None

        event.wait()
        return result




def main() -> None:
    base = "https://mcp.amap.com/sse?key=" + os.environ["AMAP_KEY"]

    sse_client = SSEClient(url=base)
    sse_client.start()
    endpoint = sse_client.wait_ready_for_endpoint()

    mcp = McpClient(
        sse=sse_client, endpoint=urljoin(base, endpoint), session=requests.Session()
    )
    mcp.initialize()
    print("init Ok!")
    tools = mcp.get_tools()
    for tool in tools['tools']:
        print(f"tool: {tool['name']}")
        print(f"  description: {tool['description']}")
        print(f"  parameters: {json.dumps(tool['inputSchema'], indent=2, ensure_ascii=False)}")
    
    print("get tools Ok!")
    
    # Call a tool
    result = mcp.call_tool("maps_weather", {"city": "北京"})
    print("call_tool Ok!")
    print(f"result: {json.dumps(result, indent=2, ensure_ascii=False)}")
    
    exit()


if __name__ == "__main__":
    main()

参考

1. RFC Streamable HTTP
2. MCP开始支持Streamable
3. 如何在CloudflareWorker 上部署 MCP
4. MCP Server

[reatang]

mermaid 画的可太好了

[LuYanFCP]

mermaid 画的可太好了

嘿嘿随便用，我就是用来写博客的