MCP协议：智能体的标准化工具接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

┌─────────────────────────────┐
│           Host (Agent)      │
│  ┌────────┐  ┌────────┐     │
│  │Client A│  │Client B│ ... │
│  └───┬────┘  └───┬────┘     │
└──────┼───────────┼──────────┘
       │           │
   stdio/SSE   stdio/SSE
       │           │
   ┌───▼────┐  ┌───▼────┐
   │Server A│  │Server B│
   │ 文件系统 │  │ 数据库  │
   └────────┘  └────────┘

1
2
3
4
5
6
7
8
9

{
  "jsonrpc": "2.0",
  "id": 42,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {"path": "/tmp/note.md"}
  }
}

1
2
3
4
5
6
7
8

{
  "jsonrpc": "2.0",
  "id": 42,
  "result": {
    "content": [{"type": "text", "text": "笔记内容..."}],
    "isError": false
  }
}

1

pip install mcp

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

# filesystem_mcp_server.py
# 文件系统 MCP Server：提供「读取文件」「列出目录」两个工具，
# 另外演示 Resources（资源）和 Prompts（提示模板）两类原语。
# 运行：python filesystem_mcp_server.py
# 依赖：pip install mcp

import os
from mcp.server.fastmcp import FastMCP

# 创建 Server 实例，名称会在握手阶段告诉 Client「我是谁」
mcp = FastMCP("filesystem-mcp")


# ---------- Tools：工具（LLM 主动调用）----------

@mcp.tool()
async def read_file(path: str) -> str:
    """读取指定路径的文本文件内容。路径必须是绝对路径。"""
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        # 用「返回文本」承载错误，而不是抛异常——
        # 这样 LLM 能看到错误描述，自行决定是否重试或换工具
        return f"错误：文件不存在 {path}"
    except Exception as e:
        return f"错误：{e}"


@mcp.tool()
async def list_directory(path: str) -> str:
    """列出指定目录下的文件与子目录名称。"""
    try:
        entries = os.listdir(path)
        lines = []
        for entry in entries:
            full = os.path.join(path, entry)
            kind = "目录" if os.path.isdir(full) else "文件"
            lines.append(f"{entry} ({kind})")
        return "\n".join(lines)
    except Exception as e:
        return f"错误：{e}"


# ---------- Resources：资源（应用层按 URI 主动读取）----------

@mcp.resource("file://{path}")
async def read_resource(path: str) -> str:
    """按 URI 读取文件内容，作为只读资源暴露给 Client。
    Client 读取 file:///tmp/note.md 时，path 会被解析为 /tmp/note.md。"""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()


# ---------- Prompts：提示模板（用户选择后由 Host 渲染）----------

@mcp.prompt()
def summarize_file(path: str) -> str:
    """生成一段提示词，让模型总结指定文件内容。"""
    return f"请阅读路径 {path} 的文件内容，并用三句话总结要点。"


# ---------- 启动 ----------

if __name__ == "__main__":
    # mcp.run() 默认以 stdio 模式启动，等待 Host 拉起并通信
    mcp.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

# mcp_client_agent.py
# 通过 MCP Client 连接文件系统 Server，并把工具接入一个简易 Agent。
# 依赖：pip install mcp openai
# 环境变量：OPENAI_API_KEY、OPENAI_BASE_URL（可选）、MODEL_NAME（可选）

import os
import sys
import asyncio
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI

# LLM 客户端：可替换为任何兼容 OpenAI 接口的模型服务
llm = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY", "sk-xxx"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
MODEL = os.getenv("MODEL_NAME", "gpt-5.4")


async def run_agent(user_query: str):
    """启动 MCP Server 子进程，建立会话，运行一轮 Agent 对话"""

    # 第 1 步：配置 stdio 连接参数，以子进程方式拉起 Server
    server_params = StdioServerParameters(
        command=sys.executable,            # 用当前 Python 解释器
        args=["filesystem_mcp_server.py"], # Server 脚本路径
        env=os.environ.copy()             # 透传环境变量
    )

    # 第 2 步：建立 stdio 连接，再创建 Client 会话
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 第 3 步：初始化握手，交换协议版本与能力声明
            await session.initialize()

            # 第 4 步：发现工具，调用 tools/list 拿到清单
            tools_resp = await session.list_tools()
            mcp_tools = tools_resp.tools
            print(f"[Client] 发现 {len(mcp_tools)} 个工具：",
                  [t.name for t in mcp_tools])

            # 第 5 步：把 MCP 工具转换为 OpenAI Function Calling 格式
            oa_tools = []
            for t in mcp_tools:
                oa_tools.append({
                    "type": "function",
                    "function": {
                        "name": t.name,
                        "description": t.description,
                        "parameters": t.inputSchema
                    }
                })

            # 第 6 步：Agent 主循环：对话 → 可能调用工具 → 回填结果 → 继续对话
            messages = [
                {"role": "system", "content": "你是一个文件助手，可以读写本地文件。"},
                {"role": "user", "content": user_query}
            ]

            for _ in range(5):  # 最多 5 轮，防止死循环
                resp = llm.chat.completions.create(
                    model=MODEL,
                    messages=messages,
                    tools=oa_tools
                )
                msg = resp.choices[0].message
                messages.append(msg.model_dump(exclude_none=True))

                # 没有工具调用，说明模型已经给出最终回答
                if not msg.tool_calls:
                    print(f"[Agent] 回复：{msg.content}")
                    break

                # 执行模型要求的每一个工具调用
                for call in msg.tool_calls:
                    fn_name = call.function.name
                    fn_args = json.loads(call.function.arguments or "{}")
                    print(f"[Agent] 调用工具 {fn_name}，参数：{fn_args}")

                    # 通过 MCP Client 调用 Server 上的工具
                    result = await session.call_tool(fn_name, fn_args)
                    # 把工具结果回填为 tool 角色消息
                    tool_text = "\n".join(
                        c.text for c in result.content if hasattr(c, "text")
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": tool_text
                    })
                    print(f"[Tool] 结果：{tool_text[:200]}")


if __name__ == "__main__":
    # 示例：让 Agent 列出 /tmp 目录并读取某个 .md 文件
    asyncio.run(run_agent("请列出 /tmp 目录下有哪些文件，并读取其中任意一个 .md 文件的内容。"))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

# weather_mcp_server.py
# 天气查询 MCP Server：暴露 get_weather 工具
# 依赖：pip install mcp httpx
# 运行：python weather_mcp_server.py

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-mcp")

# wttr.in 是免费天气服务，?format=j1 返回 JSON
WEATHER_URL = "https://wttr.in/{city}?format=j1"


@mcp.tool()
async def get_weather(city: str) -> str:
    """查询指定城市的当前天气，返回温度、天气状况、湿度、风速等。
    城市名用英文或拼音，例如 Beijing、shanghai。"""
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(WEATHER_URL.format(city=city))
            resp.raise_for_status()
            data = resp.json()

        # 解析当前天气字段
        current = data["current_condition"][0]
        temp_c = current["temp_C"]
        desc = current["weatherDesc"][0]["value"]
        humidity = current["humidity"]
        wind = current["windspeedKmph"]

        return (
            f"城市：{city}\n"
            f"温度：{temp_c}°C\n"
            f"天气：{desc}\n"
            f"湿度：{humidity}%\n"
            f"风速：{wind}km/h"
        )
    except Exception as e:
        return f"查询失败：{e}"


if __name__ == "__main__":
    mcp.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

# weather_agent.py
# 天气查询 Agent：通过 MCP 调用 weather_mcp_server
# 依赖：pip install mcp openai

import os
import sys
import asyncio
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI

llm = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY", "sk-xxx"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
MODEL = os.getenv("MODEL_NAME", "gpt-5.4")


async def run_weather_agent(user_query: str):
    # 以子进程拉起天气 Server
    server_params = StdioServerParameters(
        command=sys.executable,
        args=["weather_mcp_server.py"],
        env=os.environ.copy()
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()  # 握手

            # 发现工具并转成 OpenAI 格式
            tools_resp = await session.list_tools()
            oa_tools = [{
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.inputSchema
                }
            } for t in tools_resp.tools]

            messages = [
                {"role": "system", "content": "你是天气助手，可以查询全球城市天气。"},
                {"role": "user", "content": user_query}
            ]

            for _ in range(4):  # 最多 4 轮
                resp = llm.chat.completions.create(
                    model=MODEL, messages=messages, tools=oa_tools
                )
                msg = resp.choices[0].message
                messages.append(msg.model_dump(exclude_none=True))

                # 没有工具调用，输出最终回答并结束
                if not msg.tool_calls:
                    print(f"天气助手：{msg.content}")
                    return

                # 执行每个工具调用
                for call in msg.tool_calls:
                    fn_name = call.function.name
                    fn_args = json.loads(call.function.arguments or "{}")
                    print(f"[调用] {fn_name}({fn_args})")

                    result = await session.call_tool(fn_name, fn_args)
                    tool_text = "\n".join(
                        c.text for c in result.content if hasattr(c, "text")
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": tool_text
                    })


if __name__ == "__main__":
    asyncio.run(run_weather_agent("北京和上海今天分别多少度？需要带伞吗？"))

1
2
3
4

[调用] get_weather({'city': 'Beijing'})
[调用] get_weather({'city': 'Shanghai'})
天气助手：北京今天约28°C，晴，湿度40%；上海26°C，小雨，湿度85%。
上海有雨，建议带伞；北京晴朗，无需带伞。