AI Agent 工程实践教程 · 第 16 章

理解 MCP 如何把工具、资源和提示词统一接给 Agent，并从 Server、Tool、Resource、Prompt 和 Client 调用链路入门。

返回系列目录

第十六天：MCP 服务开发入门

前面我们已经学习了工具调用和 Agent。

工具调用解决的是：

让模型知道：什么时候该调用工具、调用哪个工具、参数是什么。

Agent 解决的是：

自动完成“模型判断 -> 调用工具 -> 把结果交回模型 -> 生成最终回答”的流程。

但是这里还有一个问题：

如果我们有很多工具，比如：

1. 查询课程。
2. 查询订单。
3. 查询学生。
4. 读取知识库文档。
5. 读取项目配置。
6. 生成某类固定提示词。

这些工具应该怎么统一接给不同的 Agent 客户端？

比如今天接 Claude Desktop，明天接 Cursor，后天接 Codex，难道每个客户端都写一套工具适配代码吗？

这就是 MCP 要解决的问题。

一句话：

MCP 是一套把工具、资源、提示词统一接给 Agent 的协议。

这里的“统一接给 Agent”要重点理解一下。

它不是说把工具、资料和提示词直接塞进大模型参数里。

它的意思是：

我们把工具、资源、提示词放到 MCP Server 里。
Agent 客户端通过 MCP 协议连接这个 Server。
连接成功后，Agent 客户端就能发现和使用这些能力。

所以 MCP 更像一个标准插座。

工具、文档、提示词不用分别为 Claude、Cursor、Codex 各写一套接入逻辑，只要按 MCP 协议暴露出来，支持 MCP 的客户端就可以连接。

1. 先用一句话理解 MCP

MCP 全称是 Model Context Protocol。

可以先不用记全称，先记住这个理解：

MCP = 给 Agent 插工具、插资料、插提示词的一套标准接口。

它不是大模型。

它也不是 Agent。

它更像一个中间层：

Agent 客户端
    |
    | MCP 协议
    v
MCP Server
    |
    v
你的工具 / 文档 / 业务系统

以前我们在 LangChain 里写工具，通常是直接把 Python 函数绑定给模型。

MCP 的思路是：

把工具放到一个独立的 MCP Server 里。
Agent 客户端通过 MCP 协议发现和调用这些工具。

这样工具就不只属于某一个 Python 程序，而是可以被支持 MCP 的客户端复用。

2. MCP Server 里能放什么

MCP Server 主要暴露三类能力。

可以先看这张图：

注意：

Prompt 不是比 Tool、Resource 更高的一层。

在 MCP 里，Tools、Resources、Prompts 是 MCP Server 并列暴露给 Agent 客户端的三类能力。

它们只是用途不同：

Tool      负责执行动作。
Resource  负责读取资料。
Prompt    负责提供可复用提示词模板。

2.1 Tools：工具

Tool 是让 Agent 做事的能力。

比如：

query_course       查询课程
get_current_time   获取当前时间
query_order        查询订单
search_knowledge   查询知识库

Tool 通常有输入参数，也有返回结果。

例如：

工具名：query_course
参数：keyword
结果：课程列表

2.2 Resources：资源

Resource 是让 Agent 读取资料的能力。

比如：

yanque://course/mcp
yanque://config/chat
yanque://doc/project-readme

Resource 更像“可读取的数据地址”。

Agent 可以根据资源 URI 读取内容。

2.3 Prompts：提示词模板

Prompt 是预设好的提示词模板。

比如：

course_study_plan
code_review_prompt
sql_analysis_prompt

它的作用是：

把常用提示词做成可复用模板，让 Agent 客户端可以按参数生成提示词。

3. 本节要做什么

这一节我们不直接写复杂业务。

我们先做一个最小 MCP 服务。

它包含：

1. 一个空 MCP Server。
2. 一个只包含 Tool 的 MCP Server。
3. 一个只包含 Resource 的 MCP Server。
4. 一个只包含 Prompt 的 MCP Server。
5. 一个完整 MCP Server。
6. 一个本地 Client，用来测试完整 MCP Server。
7. 一个接真实业务 API 的写法示例。

本节 demo 放在 AI 项目的 test/mcp_demo 下：

YanQue-AI
  └── test
      └── mcp_demo
          ├── 00_empty_server.py
          ├── 01_tool_server.py
          ├── 02_resource_server.py
          ├── 03_prompt_server.py
          ├── 04_full_server.py
          ├── 05_stdio_client.py
          ├── 06_business_api_style.py
          ├── 07_streamable_http_server.py
          ├── 08_streamable_http_client.py
          ├── 09_load_mcp_tools.py
          ├── 10_agent_with_mcp_tools.py
          └── README.md

为什么要拆这么多文件？

因为 MCP 初学时最容易卡在：

一上来既有 Tool，又有 Resource，又有 Prompt，又有 Client。

看起来代码不多，但概念会混在一起。

所以本节按下面的顺序逐步展开：

00_empty_server.py        先知道 MCP Server 怎么启动
01_tool_server.py         再知道 Tool 怎么注册
02_resource_server.py     再知道 Resource 怎么读取
03_prompt_server.py       再知道 Prompt 怎么生成
04_full_server.py         最后把三类能力合在一起
05_stdio_client.py        用 Client 测试完整调用链路
06_business_api_style.py  看真实项目里怎么接 Java 后端 API
07_streamable_http_server.py  看 MCP Server 如何用 HTTP 方式启动
08_streamable_http_client.py  看 MCP Client 如何通过 URL 连接 HTTP MCP Server
09_load_mcp_tools.py     把 MCP Tools 加载成 LangChain Tools，并直接调用
10_agent_with_mcp_tools.py  Agent 接入 MCP Server 工具的完整示例

先不要急着接真实数据库。

我们先用内存数据模拟课程。

等流程跑通以后，再把内存数据替换成：

1. HTTP API 调用。
2. 数据库查询。
3. Java 后端接口。
4. 已有 Python service。

4. 安装 MCP SDK

进入 YanQue-AI 目录：

cd /Users/huangjin/IdeaProjects/YanQue/YanQue-AI

安装 MCP Python SDK：

uv add mcp

安装完成后，pyproject.toml 里会出现类似依赖：

dependencies = [
    "mcp>=1.28.1",
]

5. 第一步：创建一个空 MCP Server

先创建文件：

test/mcp_demo/00_empty_server.py

写入最小代码：

from mcp.server.fastmcp import FastMCP


mcp = FastMCP("yanque-course-demo")


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

这里先解释三个东西。

5.1 FastMCP 是什么

FastMCP 是 Python MCP SDK 提供的快速开发工具。

可以把它理解成：

FastAPI 帮我们快速写 HTTP API。
FastMCP 帮我们快速写 MCP Server。

如果不用 FastMCP，我们要自己处理很多底层事情：

1. 解析 MCP 请求。
2. 判断客户端要调用哪个工具。
3. 校验参数。
4. 返回结果。
5. 注册资源。
6. 注册提示词。
7. 处理 stdio 通信。

用了 FastMCP 以后，我们只需要关心：

我要暴露哪些工具、资源和提示词。

5.2 transport 是什么

transport 可以理解成：

MCP Client 和 MCP Server 之间用什么方式通信。

也就是：

请求怎么发过去，结果怎么传回来。

在当前 Python MCP SDK 里，mcp.run(...) 常见写法有：

mcp.run(transport="stdio")
mcp.run(transport="streamable-http")
mcp.run(transport="sse")

它们的区别不是“工具能力不同”。

工具、资源、提示词还是那些。

区别只是：

客户端怎么连接服务端。

可以先记住这张对比表：

stdio：
    客户端启动 MCP Server 进程。
    客户端通过标准输入输出和服务端通信。
    适合本地开发、本地客户端、Claude Desktop、Cursor 这类 command + args 配置。

streamable-http：
    MCP Server 自己作为 HTTP 服务启动。
    客户端通过 URL 连接它。
    适合远程部署、团队共享、服务化接入。

sse：
    也是 HTTP 相关方式，使用 Server-Sent Events 建立消息通道。
    一些早期 MCP 客户端或旧接入方式会看到它。

5.3 stdio 是什么

这里的：

mcp.run(transport="stdio")

表示 MCP Server 使用标准输入输出通信。

它不是 Web 服务。

所以它不会显示：

http://localhost:8000

stdio 模式的意思是：

MCP 客户端启动这个 Python 进程。
客户端通过标准输入输出和它通信。

这也是本地 MCP Server 最常见的入门方式。

我们前面的 00_empty_server.py 到 06_business_api_style.py 都使用 stdio。

本地运行时，stdio 最简单：

不用开端口。
不用处理 HTTP 地址。
客户端直接启动服务端进程。

可以先用下面的方式理解：

普通命令行程序：
    人在终端输入内容
        -> 程序从标准输入 stdin 读到内容
    程序 print 输出内容
        -> 内容出现在终端屏幕上

MCP stdio 模式：
    MCP 客户端把请求写给 MCP Server
        -> MCP Server 从标准输入 stdin 读到请求
    MCP Server 把结果写出来
        -> MCP 客户端从标准输出 stdout 读到结果

先记住一句话：

stdio 就是不用 URL，而是用“输入通道”和“输出通道”让客户端和服务端说话。

在我们的 demo 里：

async with stdio_client(server_params) as (read_stream, write_stream):

这行代码主要完成两件事：

stdio_client 做了两件事：
1. 启动 04_full_server.py 这个 MCP Server。
2. 建立两条通道：
   write_stream 用来发请求。
   read_stream 用来读结果。

后面的：

async with ClientSession(read_stream, write_stream) as session:

可以理解成：

把这两条通道包装成 MCP 会话。
之后就可以用 list_tools、call_tool、read_resource、get_prompt 这些方法了。

5.4 streamable-http 是什么

stdio 的特点是：

客户端启动服务端进程。

而 streamable-http 的特点是：

服务端先自己启动，监听一个 HTTP 地址。
客户端再通过 URL 连接这个 MCP Server。

也就是说，streamable-http 更像我们平时写后端服务：

先启动服务：
    uv run python test/mcp_demo/07_streamable_http_server.py

再访问地址：
    http://127.0.0.1:8000/mcp

在代码里，核心区别是：

mcp = FastMCP(
    "yanque-http-demo",
    host="127.0.0.1",
    port=8000,
    streamable_http_path="/mcp",
)

mcp.run(transport="streamable-http")

这个示例放在：

test/mcp_demo/07_streamable_http_server.py

它和 04_full_server.py 暴露的能力类似，也包含：

1. query_course 工具。
2. get_current_time 工具。
3. yanque://course/{course_id} 资源。
4. course_study_plan 提示词。

简化代码如下：

from datetime import datetime
from typing import Any

from mcp.server.fastmcp import FastMCP


mcp = FastMCP(
    "yanque-http-demo",
    host="127.0.0.1",
    port=8000,
    streamable_http_path="/mcp",
)


COURSES: dict[str, dict[str, Any]] = {
    "langchain": {
        "id": "langchain",
        "name": "LangChain 框架实战",
        "teacher": "燕雀 AI 教研组",
        "date": "2026-07-06",
        "summary": "学习模型调用、提示词模板、工具调用和 Agent 基础流程。",
    },
    "mcp": {
        "id": "mcp",
        "name": "MCP 服务开发入门",
        "teacher": "燕雀 AI 教研组",
        "date": "2026-07-23",
        "summary": "学习 MCP Server 如何把工具、资源和提示词暴露给 Agent。",
    },
}


@mcp.tool()
def query_course(keyword: str) -> list[dict[str, Any]]:
    """按课程 ID、课程名或简介关键词查询课程。"""
    normalized_keyword = keyword.strip().lower()
    if not normalized_keyword:
        return []

    return [
        course
        for course in COURSES.values()
        if normalized_keyword in course["id"].lower()
        or normalized_keyword in course["name"].lower()
        or normalized_keyword in course["summary"].lower()
    ]


@mcp.tool()
def get_current_time() -> str:
    """返回 MCP HTTP 服务端当前时间。"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


@mcp.resource("yanque://course/{course_id}", mime_type="application/json")
def get_course_resource(course_id: str) -> dict[str, Any]:
    """读取指定课程的完整资料。"""
    course = COURSES.get(course_id)
    if course is None:
        return {
            "error": "COURSE_NOT_FOUND",
            "message": f"未找到课程：{course_id}",
        }
    return course


@mcp.prompt()
def course_study_plan(course_name: str, student_level: str = "初学者") -> str:
    """生成课程学习计划提示词。"""
    return f"请为 {student_level} 学员制定《{course_name}》学习计划。"


if __name__ == "__main__":
    mcp.run(transport="streamable-http")

启动服务端：

uv run python test/mcp_demo/07_streamable_http_server.py

启动后它监听的地址是：

http://127.0.0.1:8000/mcp

注意：

这个地址不是普通网页。

它是 MCP 客户端要连接的 HTTP MCP endpoint。

所以你用浏览器打开它，不一定会看到一个漂亮页面。

我们再写一个 HTTP 客户端连接它：

test/mcp_demo/08_streamable_http_client.py

核心代码是：

import asyncio

from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client


async def main() -> None:
    async with streamable_http_client("http://127.0.0.1:8000/mcp") as (
        read_stream,
        write_stream,
        _get_session_id,
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            tools = await session.list_tools()
            for tool in tools.tools:
                print(f"- {tool.name}: {tool.description}")

            course_result = await session.call_tool(
                "query_course",
                {"keyword": "MCP"},
            )
            print(course_result.content)


if __name__ == "__main__":
    asyncio.run(main())

这里要注意：

stdio_client(...)：
    根据 command + args 启动 MCP Server。

streamable_http_client(...)：
    不启动服务端，只连接已经启动好的 URL。

所以 HTTP 版本要开两个终端。

第一个终端启动服务端：

uv run python test/mcp_demo/07_streamable_http_server.py

第二个终端启动客户端：

uv run python test/mcp_demo/08_streamable_http_client.py

5.5 stdio 和 streamable-http 的区别

可以用下面这张表对比：

对比项              stdio                         streamable-http

谁启动服务端        客户端启动                     服务端自己先启动
怎么连接            stdin / stdout 输入输出通道    HTTP URL
有没有端口          没有                           有，例如 8000
适合哪里            本地工具、本地开发、个人使用    远程部署、团队共享、正式服务
客户端配置          command + args                 url
是否像 Web 服务     不像                           更像
浏览器能不能看      不能当网页看                   endpoint 也不是普通网页

初学和本地演示时，优先使用：

mcp.run(transport="stdio")

原因是：

配置少，最容易跑通。

如果你要把 MCP Server 部署成一个独立服务，或者让多个客户端通过网络连接，再考虑：

mcp.run(transport="streamable-http")

可以这样判断：

本地工具型 MCP：优先 stdio。
远程服务型 MCP：优先 streamable-http。
最小 demo：优先 stdio。
YanQue 业务服务化：更适合 streamable-http。

6. 第二步：添加一个 Tool

现在我们给 MCP Server 加一个工具。

这一步建议单独创建文件：

test/mcp_demo/01_tool_server.py

这个工具叫：

query_course

作用是按关键词查询课程。

完整代码变成：

from typing import Any

from mcp.server.fastmcp import FastMCP


mcp = FastMCP("yanque-course-demo")


COURSES: dict[str, dict[str, Any]] = {
    "langchain": {
        "id": "langchain",
        "name": "LangChain 框架实战",
        "teacher": "燕雀 AI 教研组",
        "date": "2026-07-06",
        "summary": "学习模型调用、提示词模板、工具调用和 Agent 基础流程。",
    },
    "mcp": {
        "id": "mcp",
        "name": "MCP 服务开发入门",
        "teacher": "燕雀 AI 教研组",
        "date": "2026-07-23",
        "summary": "学习 MCP Server 如何把工具、资源和提示词暴露给 Agent。",
    },
}


@mcp.tool()
def query_course(keyword: str) -> list[dict[str, Any]]:
    """按课程 ID、课程名或简介关键词查询课程。"""
    normalized_keyword = keyword.strip().lower()
    if not normalized_keyword:
        return []

    return [
        course
        for course in COURSES.values()
        if normalized_keyword in course["id"].lower()
        or normalized_keyword in course["name"].lower()
        or normalized_keyword in course["summary"].lower()
    ]


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

重点看这段：

@mcp.tool()
def query_course(keyword: str) -> list[dict[str, Any]]:
    ...

它的意思是：

把 query_course 这个 Python 函数注册成 MCP Tool。

注册以后，Agent 客户端就能看到这个工具。

函数的注释：

"""按课程 ID、课程名或简介关键词查询课程。"""

也很重要。

因为 Agent 会根据工具说明判断：

什么时候应该调用这个工具。

所以工具说明要写清楚，不要太随意。

7. 第三步：再添加一个获取时间的 Tool

再加一个更简单的工具：

from datetime import datetime


@mcp.tool()
def get_current_time() -> str:
    """返回 MCP 服务端当前时间。"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

这个工具虽然简单，但很适合说明一个概念：

大模型自己并不知道程序运行机器上的真实当前时间。
如果问题需要实时信息，就应该通过工具获取。

8. 第四步：添加一个 Resource

Tool 是“执行动作”。

Resource 更像“读取资料”。

这一步建议单独创建文件：

test/mcp_demo/02_resource_server.py

我们添加一个课程详情资源：

@mcp.resource("yanque://course/{course_id}", mime_type="application/json")
def get_course_resource(course_id: str) -> dict[str, Any]:
    """读取指定课程的完整资料。"""
    course = COURSES.get(course_id)
    if course is None:
        return {
            "error": "COURSE_NOT_FOUND",
            "message": f"未找到课程：{course_id}",
        }
    return course

这里的资源地址是：

yanque://course/{course_id}

如果客户端读取：

yanque://course/mcp

服务端就会调用：

get_course_resource(course_id="mcp")

然后返回 MCP 课程详情。

Resource 适合放这些内容：

1. 项目说明文档。
2. 数据库表结构。
3. 课程详情。
4. 配置信息。
5. 知识库片段。

9. 第五步：添加一个 Prompt

Prompt 是提示词模板。

这一步建议单独创建文件：

test/mcp_demo/03_prompt_server.py

我们添加一个生成课程学习计划的提示词：

@mcp.prompt()
def course_study_plan(course_name: str, student_level: str = "初学者") -> str:
    """生成课程学习计划提示词。"""
    return f"""
你是燕雀 AI 课程助教，请为 {student_level} 学员制定《{course_name}》学习计划。

要求：
1. 先用一句话说明这门课适合解决什么问题。
2. 按课前、课中、课后三个阶段安排学习任务。
3. 给出 3 个可实践的小练习。
4. 最后提醒学员哪些地方容易踩坑。
""".strip()

这个函数不是直接调用模型。

它只是返回一段提示词。

Agent 客户端拿到这段提示词以后，可以再交给模型使用。

Prompt 适合沉淀固定任务模板。

比如：

生成学习计划
代码审查
SQL 分析
面试复盘
日志诊断

10. 第六步：写一个本地 Client 测试它

只写 MCP Server 不容易看效果。

前面 01_tool_server.py、02_resource_server.py、03_prompt_server.py 是为了分开理解概念。

真正测试完整链路时，可以把它们合成一个完整服务端：

test/mcp_demo/04_full_server.py

然后再写一个客户端文件：

test/mcp_demo/05_stdio_client.py

代码如下：

import asyncio

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


async def main() -> None:
    server_params = StdioServerParameters(
        command="uv",
        args=["run", "python", "test/mcp_demo/04_full_server.py"],
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        # stdio_client 会启动上面的命令，并拿到服务端进程的输入输出通道。
        # write_stream：往 MCP Server 的 stdin 写请求。
        # read_stream：从 MCP Server 的 stdout 读响应。
        async with ClientSession(read_stream, write_stream) as session:
            # ClientSession 把 stdin/stdout 这种底层读写，封装成 list_tools、call_tool 等方法。
            await session.initialize()

            print("1. 服务端暴露的工具：")
            tools = await session.list_tools()
            for tool in tools.tools:
                print(f"- {tool.name}: {tool.description}")
            print()

            print("2. 调用工具 query_course：")
            course_result = await session.call_tool(
                "query_course",
                {"keyword": "MCP"},
            )
            print(course_result.content)
            print()

            print("3. 调用工具 get_current_time：")
            time_result = await session.call_tool("get_current_time", {})
            print(time_result.content)
            print()

            print("4. 读取资源 yanque://course/mcp：")
            resource_result = await session.read_resource("yanque://course/mcp")
            for content in resource_result.contents:
                print(content.text)
            print()

            print("5. 获取提示词 course_study_plan：")
            prompt_result = await session.get_prompt(
                "course_study_plan",
                {
                    "course_name": "MCP 服务开发入门",
                    "student_level": "Java 后端工程师",
                },
            )
            for message in prompt_result.messages:
                print(message.content.text)


if __name__ == "__main__":
    asyncio.run(main())

这段代码做了四件事：

1. 启动 MCP Server。
2. 查询服务端有哪些工具。
3. 调用 query_course 工具。
4. 调用 get_current_time 工具。
5. 读取课程资源。
6. 获取课程学习计划提示词。

注意这里：

server_params = StdioServerParameters(
    command="uv",
    args=["run", "python", "test/mcp_demo/04_full_server.py"],
)

它的意思是：

客户端负责启动 MCP Server。

所以我们运行客户端时，不需要提前手动启动服务端。

整个 stdio demo 的运行关系可以画成这样：

11. 第七步：运行 demo

进入 AI 项目：

cd /Users/huangjin/IdeaProjects/YanQue/YanQue-AI

运行客户端：

uv run python test/mcp_demo/05_stdio_client.py

如果成功，会看到类似输出：

1. 服务端暴露的工具：
- query_course: 按课程 ID、课程名或简介关键词查询课程。
- get_current_time: 返回 MCP 服务端当前时间。

2. 调用工具 query_course：
...

3. 调用工具 get_current_time：
...

4. 读取资源 yanque://course/mcp：
...

5. 获取提示词 course_study_plan：
...

这就说明：

MCP Server 已经可以把 Tool、Resource、Prompt 暴露给 MCP Client。

12. 单独启动服务端为什么没反应

如果你运行：

uv run python test/mcp_demo/04_full_server.py

你可能会发现终端没有明显输出。

这是正常的。

因为它是 stdio 模式，不是 HTTP 模式。

它启动后就在等待 MCP 客户端通过标准输入输出发送消息。

所以学习阶段建议先运行：

uv run python test/mcp_demo/05_stdio_client.py

这样客户端会自动启动服务端，并演示完整调用过程。

13. 接入 MCP 客户端时怎么配置

不同客户端配置文件位置不一样，但核心配置长得差不多。

大概是：

{
  "mcpServers": {
    "yanque-course-demo": {
      "command": "uv",
      "args": [
        "run",
        "python",
        "test/mcp_demo/04_full_server.py"
      ]
    }
  }
}

如果客户端不是在 YanQue-AI 目录下启动，建议把 server 文件写成绝对路径：

{
  "mcpServers": {
    "yanque-course-demo": {
      "command": "uv",
      "args": [
        "run",
        "python",
        "/Users/huangjin/IdeaProjects/YanQue/YanQue-AI/test/mcp_demo/04_full_server.py"
      ]
    }
  }
}

14. MCP 和普通工具调用有什么区别

前面我们在 LangChain 里写过工具：

@tool
def get_student_day_schedule(date: str) -> str:
    ...

这种工具通常只在当前 Python 程序里使用。

MCP 工具则是通过 MCP Server 暴露出去：

@mcp.tool()
def query_course(keyword: str):
    ...

差别可以这样理解：

LangChain Tool：
    工具直接绑定在当前 Agent 程序里。

MCP Tool：
    工具放在独立 MCP Server 里。
    支持 MCP 的 Agent 客户端都可以连接它。

所以 MCP 更适合做统一工具入口。

比如公司可以做一个：

yanque-business-mcp-server

里面统一放：

1. 查询课程工具。
2. 查询学生工具。
3. 查询订单工具。
4. 查询知识库资源。
5. 生成运营分析提示词。

然后不同 Agent 客户端都可以接这个服务。

15. Agent 怎么接入 MCP Server

前面 05_stdio_client.py 只是普通 MCP Client。

它能证明：

MCP Client 可以连接 MCP Server，并调用工具、资源、提示词。

但真实项目里，我们更关心的是：

Agent 能不能使用 MCP Server 暴露的工具？

可以。

这里需要一个适配器：

uv add langchain-mcp-adapters

这个依赖的作用是：

把 MCP Server 暴露的 Tools 加载成 LangChain 可以使用的 Tools。

这里会用到一个类：

from langchain_mcp_adapters.client import MultiServerMCPClient

MultiServerMCPClient 可以理解成：

LangChain 连接 MCP Server 的客户端适配器。

它本身不是 Agent。

它也不负责调用大模型。

它主要做三件事：

1. 根据配置连接一个或多个 MCP Server。
2. 从 MCP Server 读取工具列表。
3. 把 MCP Tool 转成 LangChain Agent 可以使用的 Tool。

名字里的 MultiServer 表示它可以同时连接多个 MCP Server。

比如以后可以这样配置：

client = MultiServerMCPClient(
    {
        "course_server": {
            "url": "http://127.0.0.1:8000/mcp",
            "transport": "streamable_http",
        },
        "order_server": {
            "url": "http://127.0.0.1:8001/mcp",
            "transport": "streamable_http",
        },
    }
)

这段配置的意思是：

同时连接课程 MCP Server 和订单 MCP Server。
然后把两个服务里的工具一起加载出来。

本节 demo 只连接一个 MCP Server：

client = MultiServerMCPClient(
    {
        "yanque_course": {
            "url": "http://127.0.0.1:8000/mcp",
            "transport": "streamable_http",
        }
    }
)

这里的配置可以这样读：

yanque_course：
    给这个 MCP Server 起一个本地名字。

url：
    MCP Server 的 HTTP 地址。

transport：
    使用 streamable_http 方式连接。

后面的：

tools = await client.get_tools()

意思是：

连接 MCP Server，读取它暴露的 Tools，并转换成 LangChain Tools。

整体关系可以看这张图：

15.1 先加载 MCP Tools

先看文件：

test/mcp_demo/09_load_mcp_tools.py

核心代码：

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient


async def main() -> None:
    client = MultiServerMCPClient(
        {
            "yanque_course": {
                "url": "http://127.0.0.1:8000/mcp",
                "transport": "streamable_http",
            }
        }
    )

    tools = await client.get_tools()

    print("从 MCP Server 加载到的 LangChain 工具：")
    for tool in tools:
        print(f"- {tool.name}: {tool.description}")
    print()

    tool_map = {tool.name: tool for tool in tools}
    query_course = tool_map["query_course"]

    print("直接调用 MCP 工具 query_course：")
    result = await query_course.ainvoke({"keyword": "MCP"})
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

运行：

先启动 HTTP MCP Server：

uv run python test/mcp_demo/07_streamable_http_server.py

再打开另一个终端运行：

uv run python test/mcp_demo/09_load_mcp_tools.py

这一步还没有创建 Agent。

它只是证明：

MCP Server 里的 query_course、get_current_time
已经变成了 LangChain Tool。

15.2 再把 MCP Tools 交给 Agent

再看文件：

test/mcp_demo/10_agent_with_mcp_tools.py

核心流程是：

1. 连接 MCP Server。
2. 加载 MCP Tools。
3. 创建 LangChain Agent。
4. 把 MCP Tools 作为 tools 传给 Agent。
5. 用户提问后，Agent 自己决定是否调用 MCP 工具。

核心代码：

import asyncio

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

from test.langchain_demo.demo_model import build_demo_model


SYSTEM_PROMPT = (
    "你是一个燕雀课程助手。"
    "如果用户询问课程信息，优先使用 MCP 工具查询真实课程数据。"
    "回答要简洁、自然，适合初学者理解。"
)


async def main() -> None:
    client = MultiServerMCPClient(
        {
            "yanque_course": {
                "url": "http://127.0.0.1:8000/mcp",
                "transport": "streamable_http",
            }
        }
    )

    mcp_tools = await client.get_tools()

    agent = create_agent(
        model=build_demo_model(),
        tools=mcp_tools,
        system_prompt=SYSTEM_PROMPT,
    )

    result = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "帮我查一下 MCP 课程是什么时候上，课程主要讲什么？",
                }
            ]
        }
    )

    print(result["messages"][-1].content)


if __name__ == "__main__":
    asyncio.run(main())

运行：

先启动 HTTP MCP Server：

uv run python test/mcp_demo/07_streamable_http_server.py

再打开另一个终端运行：

uv run python test/mcp_demo/10_agent_with_mcp_tools.py

这个文件需要 .env 中已经配置大模型参数。

如果运行成功，会看到 Agent 先调用 MCP 工具：

工具名：query_course
参数：{"keyword": "MCP"}

然后模型根据工具结果回答：

《MCP 服务开发入门》的上课时间是 2026-07-23，主要讲 MCP Server 如何把工具、资源和提示词暴露给 Agent。

到这里，完整链路就变成了：

用户问题
  |
  v
LangChain Agent
  |
  v
MCP Tool Adapter
  |
  v
MCP Server
  |
  v
query_course 工具

这就是 Agent 接入 MCP Server 的核心方式。

16. 真实项目里怎么接 YanQue 业务

当前 demo 使用的是内存数据：

COURSES = {
    ...
}

真实项目里一般不会这么写。

可以把工具函数改成调用后端接口。

我们单独放了一个写法示例：

test/mcp_demo/06_business_api_style.py

这个文件是接真实业务的写法示例，不需要直接运行。

因为它依赖：

1. 本地 YanQue Java 后端已经启动。
2. 后端接口路径真实存在。
3. 接口鉴权、签名、登录态已经处理好。

它主要说明：

MCP Tool 里面不一定写死内存数据。
MCP Tool 可以调用已有业务系统 API。

示例代码大概是：

from typing import Any

import httpx
from mcp.server.fastmcp import FastMCP


mcp = FastMCP("yanque-business-api-demo")


YANQUE_ADMIN_BASE_URL = "http://localhost:8080"


@mcp.tool()
def query_course_from_admin(keyword: str) -> dict[str, Any]:
    """演示如何把 MCP Tool 改造成调用 YanQue Java 后端 API。"""
    response = httpx.get(
        f"{YANQUE_ADMIN_BASE_URL}/api/course/page",
        params={"keyword": keyword},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

这段代码只是一个方向。

真实落地时，不能只写一个 httpx.get 就完事。

还要补：

1. 登录态或服务端 token。
2. 请求签名。
3. 权限校验。
4. 超时设置。
5. 错误码转换。
6. 敏感字段脱敏。
7. 返回结构整理。

如果 YanQue 的业务逻辑主要在 Java Spring Boot 后端里，推荐方式是：

MCP Server
    |
    | HTTP
    v
YanQue-Admin / YanQue 后端 API
    |
    v
Service / Mapper / Database

这样 MCP Server 不需要重复写一套业务逻辑。

它只是把已有后端能力包装成 Agent 可以调用的工具。

17. 写 MCP 工具时要注意什么

MCP 看起来只是把函数暴露出去。

但真实项目里要特别注意安全。

17.1 工具参数不要太随意

比如学生查课表工具，不建议让模型传：

studentId
classId

更合理的是只让模型传：

date

当前学生是谁，应该由登录态或服务端上下文决定。

这样可以避免模型传错 ID，甚至查到别人的数据。

17.2 写操作要谨慎

查询类工具风险较低。

但如果工具会执行写操作，比如：

创建订单
删除课程
修改学生信息
发送通知

就必须增加确认和权限控制。

不要让 Agent 在没有确认的情况下直接执行高风险操作。

17.3 返回结果要结构化

尽量返回清晰结构：

{
  "id": "mcp",
  "name": "MCP 服务开发入门",
  "date": "2026-07-23"
}

不要只返回一大段混乱字符串。

结构化结果更方便 Agent 理解，也更方便后续调试。

17.4 错误信息要明确

不要只返回：

失败了

更好的错误是：

{
  "error": "COURSE_NOT_FOUND",
  "message": "未找到课程：xxx"
}

这样 Agent 才知道下一步该怎么处理。

18. 本节小结

这一节我们学习了 MCP 的最小闭环。

重点记住：

MCP Server 是给 Agent 提供能力的服务。
Tool 让 Agent 做事。
Resource 让 Agent 读资料。
Prompt 让 Agent 获取提示词模板。
FastMCP 帮我们快速创建 MCP Server。
transport 决定 MCP Client 和 MCP Server 怎么通信。
stdio 模式适合本地 MCP 客户端启动和通信。
streamable-http 模式适合把 MCP Server 当成独立 HTTP 服务部署。

本节最重要的一句话：

MCP 不是让模型变聪明，而是让 Agent 可以用标准方式连接你的业务能力。

下一步可以继续做两件事：

1. 把 query_course 从内存数据改成调用 YanQue 后端接口。
2. 把课程、订单、学生等查询能力整理成一个正式的 YanQue MCP Server。