概述

当你理解了 Agent 如何调用 Tools 后，MCP 的概念就会变得非常直观。
这是一个从零基础到实战的 MCP 完整教程。

MCP 是什么？为什么需要它？

在你之前的代码中，Tools 是和你的 Agent 代码紧密耦合的——它们都写在同一个 Python 文件中。MCP (Model Context Protocol) 要解决的核心问题就是：将 Tools 从 Agent 中解耦出来。

一个简单的比喻

• 传统方式：就像在一个厨房里，厨师（Agent）和所有厨具（Tools）都在同一个房间。厨师直接伸手就能拿到刀和锅。
• MCP 方式：厨师（Agent）在厨房，但厨具（Tools）在另一个专业的工具房（MCP Server）里。厨师需要通过一个标准化的传送带（MCP 协议）来获取和使用工具。

这样做的巨大优势：

1. 安全性：Tools 在独立的进程中运行，一个工具的崩溃不会影响整个 Agent。
2. 语言无关性：Tools 可以用任何语言编写（Python, JavaScript, Go, Rust…），只要遵循 MCP 协议。
3. 可复用性：一套 Tools 可以被多个不同的 Agent 使用。
4. 易于管理：Tools 可以独立开发、部署和更新。

理解核心概念

MCP 的四个核心组件

组件	作用	类比
MCP Server	工具提供方：独立进程，提供一系列 Tools	专业的工具房
MCP Client	工具使用方：你的 Agent 程序	厨师
SSE 协议	通信方式：基于 Server-Sent Events 的标准化协议，可以理解是数据格式，比如http协议	标准化传送带
Transport	连接方式：Stdio（标准输入输出）或 SSH，使用哪种方式传输，比如http协议传输是tcp	传送带的类型

MCP 的工作原理

你的 Agent (MCP Client)
     │
     │ 通过 Stdio/SSH 连接
     │
     ▼
MCP Server (独立进程)
     │
     │ 执行实际操作
     │
     ▼
文件系统/数据库/API/其他资源

环境搭建与初体验

1.uv安装

简介

uv 是由 Astral 公司（也是开发了超快 Python 格式化工具 Ruff 的公司）开发的新工具。它的核心目标是解决 Python 包管理和项目工作流中的速度与复杂度问题。

它的主要功能包括：

• 包安装器：替代 pip 和 pip-tools，安装包的速度极快。
• 虚拟环境管理：替代 virtualenv 或 venv，可以快速创建和管理虚拟环境。
• 依赖解析器：替代 pip-compile，快速生成锁定的依赖文件。
• 项目初始化：快速创建新项目。

安装

choco install uv   #建议全局安全
pip install uv   #需要安装全局python环境

项目环境

创建mcp项目

# 1. 创建项目文件夹并进入
mkdir mcp-tutorial && cd mcp-tutorial

# 2. 用uv初始化Python项目（自动创建虚拟环境）
uv init

# 3. 安装MCP SDK（带CLI工具，用于运行/调试服务器）
uv add "mcp[cli]"

如果是已经存在的项目 直接uv init 会创建pyproject.toml,注意

requires-python = ">=3.10"

默认是读取当前python环境的版本 但是mcp开发要求3.10以上的版本，所有手动改下版本，他会自动安装.venv文件在本地项目下管理环境，在uv init之后立马改版本，改好后在uv add。
添加包后，在pycharm设置使用当前项目的.venv环境，如果existing有就选择，没有就选择.venv/Scripts/python.exe即可

2.运行你的第一个 MCP Server

helloworld

MCP 提供了一些官方示例服务器，我们来体验一下，创建basic_mcp_server.py：

"""
FastMCP quickstart example.

cd to the `examples/snippets/clients` directory and run:
    uv run server fastmcp_quickstart stdio
"""

from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("Demo")


# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b


# Add a dynamic greeting resource template
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"
# 关键修改：URI 无动态参数（固定为 greeting://default）
@mcp.resource("greeting://default")
def get_greeting() -> str:  # 无需 name 参数
    """Get a static greeting"""
    return "Hello, this is a static resource!"

# Add a prompt
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    """Generate a greeting prompt"""
    styles = {
        "friendly": "Please write a warm, friendly greeting",
        "formal": "Please write a formal, professional greeting",
        "casual": "Please write a casual, relaxed greeting",
    }

    return f"{styles.get(style, styles['friendly'])} for someone named {name}."
# 4. 运行服务器
if __name__ == "__main__":
    mcp.run(transport="stdio")

这会启动一个提供简单的工具（加法），一个restful资源，一个提示词。保持这个终端运行。

uv run mcp dev basic_mcp_server.py

此时会自动打开MCP Inspector 调试网页

MCP 能力

mcpserver能力介绍:

1. Resource :是 MCP 中最基本的概念，代表一个具体的、可读的数据实体。你可以把它想象成一个文件、一个数据库查询结果、一个 API 响应或任何一段可供 AI 读取和引用的信息,比如例子中的@mcp.resource(“greeting://default”)没有任何参数就是一个资源。
2. ResourceTemplate ：是一个用于发现和生成 Resource 的蓝图或模式。它本身不是资源，而是定义了如何找到或创建一类资源的规则，就是通过参数获取一个资源，比如例子中的@mcp.resource(“greeting://{name}”)。
3. Prompt 是 MCPServer 向 AI 客户端注册的预定义提示词模板。它允许服务器为用户提供一些“开箱即用”的、功能强大的对话起点，而不是让用户每次都从零开始编写复杂的提示词，Prompt 是传入一组参数（可能包括关键字），然后返回一个填充好的、完整的、可直接使用的提示词，比如代码中的@mcp.prompt() greet_user 。
4. Tools（也称为 Function Calling）允许 AI 执行操作而不仅仅是读取数据。它赋予了 AI 与外部世界交互和写的能力。

MCP Inspector

MCP Inspector 是一个官方开发的、用于可视化调试和监控 MCP（Model Context Protocol）连接和通信的图形化工具。

您可以把它理解为一个 MCP 的 “开发者调试面板” 或 “网络监视器”。

为什么需要它？它能解决什么问题？

在没有 Inspector 之前，调试 MCP Server 非常困难，你只能通过查看终端里打印的日志来猜测 JSON-RPC 消息的往来，效率极低。

MCP Inspector 通过可视化界面完美解决了这个问题：

1. 服务器连接与管理

让你可以轻松地添加、配置、启动和停止一个或多个 MCP Server。你不需要手动写配置文件和通过命令行启动，只需在界面点击即可。

2. 实时消息流监控

这是最核心的功能。Inspector 会实时显示 Client 和 Server 之间所有的 JSON-RPC 请求和响应，像一个“抓包工具”。

• 你可以看到每条消息的详细内容（方法名、参数、结果、错误）。
• 消息通常用不同颜色标记（如请求、成功响应、错误）。
• 这让你一眼就能看出通信是否成功，参数是否正确，结果是否符合预期。

3. 可视化浏览 Server 能力

Inspector 会自动列出已连接 Server 提供的所有能力（Capabilities）：

• Resources & Templates：以文件树等友好形式展示所有可用的资源和模板，你可以直接点击查看内容。
• Prompts：列出所有注册的提示词模板，你可以直接填写参数并测试运行，看会生成什么样的最终提示词。
• Tools：列出所有可用的工具，你可以直接填写参数并调用它们，实时查看执行结果和返回内容。

4. 错误诊断

当通信出现错误（例如 Server 崩溃、返回了非法格式的 JSON、权限错误等）时，Inspector 会清晰地展示错误信息，帮助你快速定位问题所在。

MCP Inspector 本质上就是一个专门的、功能强大的客户端（Client）。 它存在的唯一目的就是去连接、测试和调试其他的 MCP Server

通过上面的helloworld例子打开了，可以看到左侧的Connect点击连接到mcp server

可以看到连接后右侧看到resource prompt tool

点击对应的List按钮可以查看对应的能力

点击我们列表的资源模板，输入name参数点击Read Resource

可以看到返回

同理可测试prompt和tools

3.完整 MCP Server/Client

这里不使用之前的llama_index，使用这个例子完整演示自己用prompot告诉ai 有哪些工具参数是什么让他自己调用，之前llama_index会帮我们自动生成。
本示例包含两部分：

1. MCP 服务端：提供「加法工具」「天气查询工具」和「用户偏好资源模板」；
2. MCP 客户端：连接服务端，通过 Ollama Llama3 模型分析用户需求，自动决策调用哪个工具 / 资源，最终返回结果。

三种 transport 模式

1️⃣"stdio"

• 全称：Standard Input / Standard Output

• 原理：客户端和服务端在本地通过 子进程的 stdin/stdout 通信。

• 特点：

  • 适合 本地调试 或 单机运行。
  • 启动客户端时，MCP 服务端是作为子进程被启动的。
  • 速度快，延迟低。

• 限制：

  • 不能跨机器访问。
  • 客户端必须能启动服务端进程。

• 适用场景：本地开发，快速验证工具/资源逻辑。

---

2️⃣ "sse"

• 全称：Server-Sent Events

• 原理：服务端通过 HTTP 单向推送（SSE）把消息发送给客户端。

• 特点：

  • 是 单向通信：服务端可以推送消息到客户端，客户端一般不能发请求改变状态。
  • 支持 实时流式输出（比如生成式模型的 token 流）。

• 限制：

  • 不能做完整双向请求/响应。
  • 需要客户端额外逻辑处理 SSE 流。

• 适用场景：

  • 想要 实时流式生成结果，但客户端本身不需要频繁调用 MCP 工具。

---

3️⃣ "streamable-http"

• 全称：Streamable HTTP Transport

• 原理：通过 HTTP 双向流式连接（WebSocket 或长轮询）进行通信。

• 特点：

  • 双向通信：客户端可以调用工具、读取资源，服务端也能推送流式结果。
  • 支持 流式生成（token 或事件）+ 完整请求/响应。
  • 可以跨机器访问（远程调用）。

• 限制：

  • 比 stdio 稍慢，涉及网络开销。
  • 需要服务端通过 transport="streamable-http" 启动。

• 适用场景：

  • 远程调用 MCP 服务
  • 需要流式生成或实时反馈
  • 生产环境部署

编写 MCP 服务端（提供工具和资源）

以下例子使用stdio编写
创建文件 mcp_server.py，定义工具（加法、天气查询）和资源模板（用户偏好）：

from mcp.server.fastmcp import FastMCP, Context
from typing import Optional

# 1. 初始化 MCP 服务端（用 HTTP 传输，方便客户端远程连接）
mcp = FastMCP(name="FunctionalMcpServer")

# ------------------------------
# 2. 定义「工具（Tool）」：加法计算
# ------------------------------
@mcp.tool()
def add(a: int, b: int, ctx: Optional[Context] = None) -> int:
    """
    工具描述（Llama3 会读取此说明判断用途）：
    Add two integers and return the sum.
    Args:
        a: First integer to add
        b: Second integer to add
    Returns: Sum of a and b
    """
    if ctx:
        ctx.debug(f"Calling add tool: a={a}, b={b}")
    return a + b

# ------------------------------
# 3. 定义「工具（Tool）」：天气查询（模拟真实接口）
# ------------------------------
@mcp.tool()
def get_weather(city: str, ctx: Optional[Context] = None) -> str:
    """
    工具描述（Llama3 会读取此说明判断用途）：
    Get simulated weather information for a given city.
    Args:
        city: Name of the city to query (e.g., "Beijing", "Shanghai")
    Returns: Simulated weather string
    """
    # 模拟天气数据（实际场景可调用第三方天气API）
    weather_db = {
        "Beijing": "Sunny, 25°C, light wind",
        "Shanghai": "Cloudy, 22°C, no wind",
        "Guangzhou": "Rainy, 28°C, moderate wind",
        "Shenzhen": "Sunny, 26°C, light wind"
    }
    result = weather_db.get(city, f"Weather data for {city} not found")
    if ctx:
        ctx.debug(f"Calling get_weather tool: city={city}, result={result}")
    return result

# ------------------------------
# 4. 定义「资源模板（Resource Template）」：用户偏好
# ------------------------------
@mcp.resource("user://{user_id}/preference")  # 动态参数：user_id（用户ID）
def get_user_preference(ctx: Context, user_id: str) -> dict:
    """
    资源描述（Llama3 会读取此说明判断用途）：
    Get personalized preference of a user (e.g., favorite city, temperature unit).
    Args:
        user_id: Unique ID of the user (e.g., "user_001", "user_002")
    Returns: User preference dictionary
    """
    # 模拟用户偏好数据（实际场景可从数据库读取）
    user_preferences = {
        "user_001": {"favorite_city": "Beijing", "temp_unit": "°C"},
        "user_002": {"favorite_city": "Shanghai", "temp_unit": "°F"},
        "user_003": {"favorite_city": "Shenzhen", "temp_unit": "°C"}
    }
    preference = user_preferences.get(user_id, {"error": f"User {user_id} not found"})
    ctx.debug(f"Fetching resource: user://{user_id}/preference, result={preference}")
    return preference

# ------------------------------
# 5. 启动服务端（HTTP 传输，端口 8000）
# ------------------------------
if __name__ == "__main__":
    mcp.run(
        transport="stdio",  # 客户端使用stdio连接
    )

编写 MCP 客户端agent

创建文件 mcp_client.py，首先连接服务器获取所有工具，和资源模板，使用ollam调用llama3来对提出的用户问题进行分析规划（进行决策调用）返回调用的类型和工具名称以及参数，agent根据返沪i调用远程：

import asyncio
import os
import json
import re
from pydantic import AnyUrl
import httpx
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.shared.context import RequestContext
from mcp import types



# ------------------------------
# 配置参数
# ------------------------------
# MCP服务端启动参数（与官网示例保持一致）
SERVER_PARAMS = StdioServerParameters(
    command="uv",
    args=["run", "mcp_server.py"],  # 假设服务端文件名为mcp_server.py
    env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
)

OLLAMA_API_URL = "http://localhost:11434/v1"
API_KEY = "llama3"
LLAMA_MODEL = "llama3"

client = OpenAI(
    base_url=OLLAMA_API_URL,  # Ollama API 地址
    api_key=API_KEY  # Ollama 默认无需真实 API Key，填任意值即可
)

# ------------------------------
# Ollama Llama3决策逻辑
# ------------------------------
async def get_llama_mcp_command(user_query: str, tools_meta: list, resourceTemplates: list) -> dict:
    """通过Llama3模型分析用户需求，生成MCP调用指令"""
    prompt = f"""
    你是MCP指令生成器，需要分析用户查询并决定：
    1. 是否调用工具(Tool)或读取资源(Resource)
    2. 具体调用哪个工具/资源（参考下方元数据）
    3. 需要传递哪些参数（必须匹配工具/资源的定义）

    --- MCP服务端工具元数据 ---
    {json.dumps([{"name": t.name, "description": t.description, "parameters": t.inputSchema} for t in tools_meta], indent=2)}

    --- MCP服务端资源元数据 ---
    {json.dumps([{"uri": r.uriTemplate, "description": r.description} for r in resourceTemplates], indent=2)}

    --- 用户查询 ---
    {user_query}

    --- 输出要求 ---
    只返回JSON对象，不需要额外文本，返回的首字母一定是{{尾字幕一定是:}}。JSON必须而且只能包含：
    - "type": "tool"（调用工具）或 "resource"（读取资源）
    - "name": 工具名称（tool类型）或资源URI（resource类型）
    - "params": 参数字典（例如add工具的{{"a":3, "b":5}}）
    - "user_id": 默认为"user_001"
    """
    response = client.chat.completions.create(
        model=LLAMA_MODEL,  # 指定模型
        messages=[
            {"role": "system", "content": "你是一个mcp助手，只能输出有效的JSON，不要额外解释。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0,  # 控制生成多样性
        max_tokens=512,  # 最大生成 token 数
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)


# ------------------------------
# 采样回调（官网示例要求）
# ------------------------------
async def handle_sampling_message(
        context: RequestContext[ClientSession, None],
        params: types.CreateMessageRequestParams
) -> types.CreateMessageResult:
    """可选的采样回调函数，官网示例要求必须提供"""
    return types.CreateMessageResult(
        role="assistant",
        content=types.TextContent(type="text", text="正在处理你的请求..."),
        model="llama3:8b",
        stopReason="endTurn",
    )


# ------------------------------
# 客户端主逻辑
# ------------------------------
async def run(user_query: str):
    # 1. 通过stdio连接MCP服务端（官网标准方式）
    async with (stdio_client(SERVER_PARAMS) as (read, write)):
        # 2. 创建客户端会话（必须传入采样回调）
        async with ClientSession(
                read,
                write,
                sampling_callback=handle_sampling_message
        ) as session:
            # 3. 初始化连接（官网示例必备步骤）
            await session.initialize()

            # 4. 获取服务端元数据（工具、资源、提示词）
            tools = await session.list_tools()
            resourceTemplates = await session.list_resource_templates()
            prompts = await session.list_prompts()

            print(f"[服务端信息] 工具: {[t.name for t in tools.tools]}")
            print(f"[服务端信息] 资源: {[r.uriTemplate for r in resourceTemplates.resourceTemplates]}")
            print(f"[服务端信息] 提示词: {[p.name for p in prompts.prompts]}\n")

            # 5. 调用Llama3生成MCP指令
            print(f"[用户查询] {user_query}")
            print(f"[Llama3] 正在分析查询...\n")

            mcp_command = await get_llama_mcp_command(
                user_query,
                tools.tools,
                resourceTemplates.resourceTemplates
            )
            print(f"[Llama3决策结果]\n{json.dumps(mcp_command, indent=2)}\n")

            # 6. 执行MCP调用
            result = None
            if mcp_command["type"] == "tool":
                # 调用工具（官网示例格式）
                result = await session.call_tool(
                    mcp_command["name"],
                    arguments=mcp_command["params"]
                )
                # 解析工具返回结果
                if isinstance(result.content[0], types.TextContent):
                    result = result.content[0].text
                else:
                    result = result.structuredContent
            elif mcp_command["type"] == "resource":
                # 读取资源（官网示例格式，使用AnyUrl）
                name = resolve_name(mcp_command, "name")
                resource_content = await session.read_resource(AnyUrl(name))
                # 解析资源返回结果
                if isinstance(resource_content.contents[0], types.TextContent) or isinstance(resource_content.contents[0], types.TextResourceContents) :
                    result = resource_content.contents[0].text
            print(f"[最终结果] {result}")
            return result

def resolve_name(cmd: dict,key: str) -> str:
    """
    替换name中的占位符 {xxx}，优先取 params 里的值，否则取顶层字段（如 user_id）
    """
    name = cmd[key]
    params = cmd.get("params", {})
    # 占位符格式 {xxx}
    pattern = re.compile(r"\{(\w+)\}")

    def replacer(match):
        key = match.group(1)
        # 优先用 params，其次顶层字段
        return str(params.get(key) or cmd.get(key) or f"{{{key}}}")

    return pattern.sub(replacer, name)
# ------------------------------
# 入口函数
# ------------------------------
def main():
    # 测试查询（可替换为其他需求）
    test_queries = [
        "计算15加23的结果",
        "查询北京的天气",
        "获取user_002的偏好设置"
    ]
    # 运行第一个测试查询
    asyncio.run(run(test_queries[2]))


if __name__ == "__main__":
    main()

运行结果
使用stdio运行结果只需要运行client自动stdio启动server.

C:\Users\admin\PycharmProjects\pythonProject\.venv\Scripts\python.exe C:\Users\admin\PycharmProjects\pythonProject\mcp\http\mcp_client.py 
[服务端信息] 工具: ['add', 'get_weather']
[服务端信息] 资源: ['user://{user_id}/preference']
[服务端信息] 提示词: []

[用户查询] 获取user_002的偏好设置
[Llama3] 正在分析查询...

[Llama3决策结果]
{
  "type": "resource",
  "name": "user://{user_id}/preference",
  "params": {
    "user_id": "user_002"
  }
}

[最终结果] {
  "favorite_city": "Shanghai",
  "temp_unit": "°F"
}

使用http连接

mcp_server.py脚本修改

#如果需要修改端口可以修改FastMCP定义
mcp = FastMCP(name="FunctionalMcpServer",port=8000,host="0.0.0.0")

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",  # 用 HTTP 协议
        mount_path="/mcp",  # MCP 协议的挂载点
    )

http方式要求先启动服务端，直接run即可。
启用会默认8000端口，路径是/mcp,调用路径就是http://localhost:8000/mcp

C:\Users\admin\PycharmProjects\pythonProject\.venv\Scripts\python.exe C:\Users\admin\PycharmProjects\pythonProject\mcp\http\mcp_server.py 
INFO:     Started server process [30572]
INFO:     Waiting for application startup.
[09/22/25 15:20:50] INFO     StreamableHTTP      streamable_http_manager.py:110
                             session manager                                   
                             started                                           
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

mcp_client.py脚本修改

#定义全局变量 服务器地址
MCP_SERVER_URL = "http://localhost:8000/mcp"
#修改服务器连接方式stdio_client->streamablehttp_client
async def run(user_query: str):
    # 1. 通过stdio连接MCP服务端（官网标准方式）
    async with (streamablehttp_client(MCP_SERVER_URL) as (read, write, _get_session_id)):
        # 2. 创建客户端会话（必须传入采样回调）
        async with ClientSession(
                read,
                write,
                sampling_callback=handle_sampling_message

运行客户端，会自动连接server并规划调用。

使用ui客户端连接

这里安装一个ai客户端：cherrystudio。
运行刚刚写的streamable-http版本mcp_server.py，假设开放的路径是：http://localhost:8999/mcp
打开cherrystuido点右上角的设置

点击添加-》快速创建-》类型选择=可流式传输的http

输入URL

点击启动

回到首页点击助手，可以新建一个助手或者默认助手，点击开启mcp

点击图片上的你好我是默认助手 或者在默认助手右键编辑

点击mcp服务器，启用mcp

输入问题，比如 北京天气是：

可以看到调用了get_weather的天气，输出的就是mcp_server.py里的定义

再来看看加法计算公式

模拟ui回答调用工具

上面这个ui客户端cherrystuido既能回答问题，也能通过内容判断是否调用工具，这是如何做到了，其实就是在prompt中添加决策规则描述，因为返回的是json，可以在json最外层让模型给一个direct_answer文本回复，文本中放置一些{tools1}这样的占位符，actions告诉我要调用哪些工具，我agent调用，根据根据名称将direct_answer的占位符给替换成结果即可，以下给一个完整的客户端

import asyncio
import os
import json
import re
from pydantic import AnyUrl
import httpx
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.streamable_http import streamablehttp_client
from mcp.shared.context import RequestContext
from mcp import types

# ------------------------------
# 配置参数
# ------------------------------

MCP_SERVER_URL = "http://localhost:8000/mcp"
# OLLAMA_API_URL = "http://localhost:11434/v1"
# API_KEY = "llama3"
# LLAMA_MODEL = "llama3"
OLLAMA_API_URL = "https://open.bigmodel.cn/api/paas/v4/"
API_KEY = "xxxxxxxxxxxxxxxxxxx"
LLAMA_MODEL = "glm-4.5"

client = OpenAI(
    base_url=OLLAMA_API_URL,
    api_key=API_KEY
)


# ------------------------------
# 改进的Llama3决策逻辑
# ------------------------------
async def get_llama_mcp_command(user_query: str, tools_meta: list, resourceTemplates: list) -> dict:
    """通过Llama3模型分析用户需求，决定是直接回答还是调用工具"""

    # 构建工具和资源的描述
    tools_description = "\n".join([
        f"- {tool.name}: {tool.description}"
        for tool in tools_meta
    ])

    resources_description = "\n".join([
        f"- {resource.uriTemplate}: {resource.description}"
        for resource in resourceTemplates
    ])

    prompt = f"""
请分析用户查询，决定是否需要调用工具或直接回答。

可用工具：
{tools_description}

可用资源：
{resources_description}

用户查询："{user_query}"

决策规则：
1. 如果用户查询需要实时数据、计算、查询等工具能力，选择调用工具,一个问题可能有多个工具调用返回json数组
2. 如果只是普通对话、自我介绍、知识问答等，选择直接回答
3. 如果工具列表为空，只能直接回答
4. 每个json对象一个唯一name。
5. 每次返回都一定要有一个direct_answer，direct_answer中可以用{{工具name,比如c_1}}来占位工具返回 我调用工具会自动替换，最终我会输出这个direct_answer
   注意direct_answer中的内容尽量随机，回答友好。

请返回JSON格式：
- 最终json格式返回如下：
   {{
       "actions":[
        {{
        "action": "call_tool",
        "tool_name": "工具名称",
        "name":"c_1",
        "parameters": {{"参数名": "参数值"}}
        }}
       ]
       "direct_answer": "你想计算xx和某某的和，结果:{{c_1}},你想查询天气：beijing，天气:{{action中的某个name}}"
   }}
   注意占位符一定是actions中的action的name
- 如果需要调用工具：
  [{{
    "action": "call_tool",
    "tool_name": "工具名称",
    "name":"r_1"
    "parameters": {{"参数名": "参数值"}}
  }}]
- 需要读取资源或者资源模板，注意如果模板resource_uri中有占位符需要提取参数传入：
  [{{
    "action": "read_resource", 
    "resource_uri": "user://{{user_id}}/preference",
    "name":"r_1"
    "parameters": {{"user_id": "user003"}}
  }},
  {{
    "action": "read_resource", 
    "resource_uri": "user://001",
    "name":"r_1"
  }}
  ]
- 如果直接回答：
  {{
     "direct_answer": "我是一个超级计算机"
  }}
"""

    try:
        response = client.chat.completions.create(
            model=LLAMA_MODEL,
            messages=[
                {"role": "system",
                 "content": "你是一个智能助手，需要根据用户查询决定是否调用工具。请严格返回有效的JSON格式。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.5,
            max_tokens=1024,
            response_format={"type": "json_object"}
        )

        result = json.loads(response.choices[0].message.content)
        return result

    except Exception as e:
        print(f"模型调用失败: {e}")
        # 默认直接回答
        return {
            "action": "direct_answer",
            "response": "抱歉，我暂时无法处理这个请求。"
        }


# ------------------------------
# 采样回调
# ------------------------------
async def handle_sampling_message(
        context: RequestContext[ClientSession, None],
        params: types.CreateMessageRequestParams
) -> types.CreateMessageResult:
    return types.CreateMessageResult(
        role="assistant",
        content=types.TextContent(type="text", text="正在处理你的请求..."),
        model="llama3:8b",
        stopReason="endTurn",
    )


# ------------------------------
# 改进的客户端主逻辑
# ------------------------------
async def run(user_query: str):
    try:
        # 1. 连接MCP服务端
        async with streamablehttp_client(MCP_SERVER_URL) as (read, write, _get_session_id):
            # 2. 创建客户端会话
            async with ClientSession(
                    read,
                    write,
                    sampling_callback=handle_sampling_message
            ) as session:
                # 3. 初始化连接
                await session.initialize()

                # 4. 获取服务端元数据
                tools = await session.list_tools()
                resourceTemplates = await session.list_resource_templates()
                prompts = await session.list_prompts()

                print(f"[服务端信息] 工具: {[t.name for t in tools.tools]}")
                print(f"[服务端信息] 资源: {[r.uriTemplate for r in resourceTemplates.resourceTemplates]}")
                print(f"[用户查询] {user_query}")

                # 5. 调用Llama3进行决策
                print("[Llama3] 正在分析查询...")
                decisions = await get_llama_mcp_command(
                    user_query,
                    tools.tools,
                    resourceTemplates.resourceTemplates
                )

                print(f"[Llama3决策结果]\n{json.dumps(decisions, indent=2)}")

                # 6. 根据决策执行不同操作
                result = None
                tool_return = {}
                for decision in decisions["actions"]:
                    action = decision.get("action")

                    if action == "call_tool" and decision.get("tool_name"):
                        # 调用工具
                        tool_name = decision["tool_name"]
                        name = decision["name"]
                        parameters = decision.get("parameters", {})

                        print(f"[执行] 调用工具: {tool_name}, 参数: {parameters}")
                        tool_result = await session.call_tool(tool_name, arguments=parameters)

                        # 解析工具结果
                        if tool_result.content:
                            if isinstance(tool_result.content[0], types.TextContent):
                                result = tool_result.content[0].text
                            else:
                                result = str(tool_result.content)
                        else:
                            result = "工具调用成功，但无返回内容"
                        tool_return[name]=result

                    elif action == "read_resource" and decision.get("resource_uri"):
                        # 读取资源
                        resource_uri = decision["resource_uri"]
                        name = decision["name"]
                        # 解析URI中的占位符
                        resolved_uri = resolve_uri(resource_uri, decision.get("parameters", {}))

                        print(f"[执行] 读取资源: {resolved_uri}")
                        resource_content = await session.read_resource(AnyUrl(resolved_uri))

                        # 解析资源结果
                        if resource_content.contents:
                            content = resource_content.contents[0]
                            if hasattr(content, 'text'):
                                result = content.text
                            else:
                                result = str(content)
                        else:
                            result = "资源读取成功，但无内容"
                        tool_return[name] = result
                    else:
                        result = "决策格式错误，无法处理请求。"
                direct_answer=decisions.get("direct_answer")
                # print(f"[原始最终回答内容] {direct_answer}")
                # print(f"[工具返回结果] {tool_return}")
                if direct_answer:
                    result = direct_answer.format(**tool_return)
                print(f"[最终结果] {result}")
                return result

    except Exception as e:
        print(f"[错误] 处理过程中发生异常: {e}")
        return f"处理失败: {str(e)}"


def resolve_uri(uri_template: str, params: dict) -> str:
    """解析资源URI模板中的占位符"""

    def replace_placeholder(match):
        key = match.group(1)
        return str(params.get(key, f"{{{key}}}"))

    return re.sub(r'\{(\w+)\}', replace_placeholder, uri_template)


# ------------------------------
# 测试函数
# ------------------------------
async def test_queries():
    test_cases = [
        # "计算15加23的结果",  # 应该调用计算工具
        # "查询北京的天气",  # 应该调用天气工具
        "获取user_002的偏好设置，同时计算3344545加3444结果，并且介绍下你自己",  # 应该读取资源
        # "请介绍下你自己",  # 应该直接回答
        # "今天星期几？",  # 应该直接回答（如果无相关工具）
    ]

    for i, query in enumerate(test_cases, 1):
        print(f"\n{'=' * 50}")
        print(f"测试用例 {i}: {query}")
        print('=' * 50)

        result = await run(query)
        print(f"测试结果: {result}")


def main():
    asyncio.run(test_queries())


if __name__ == "__main__":
    main()

运行结果

==================================================
测试用例 1: 获取user_002的偏好设置，同时计算3344545加3444结果，并且介绍下你自己
==================================================
[服务端信息] 工具: ['add', 'get_weather']
[服务端信息] 资源: ['user://{user_id}/preference']
[用户查询] 获取user_002的偏好设置，同时计算3344545加3444结果，并且介绍下你自己
[Llama3] 正在分析查询...
[Llama3决策结果]
{
  "actions": [
    {
      "action": "read_resource",
      "resource_uri": "user://{user_id}/preference",
      "name": "r_1",
      "parameters": {
        "user_id": "user_002"
      }
    },
    {
      "action": "call_tool",
      "tool_name": "add",
      "name": "c_1",
      "parameters": {
        "a": 3344545,
        "b": 3444
      }
    }
  ],
  "direct_answer": "\u60a8\u597d\uff01\u6211\u662f\u4e00\u4e2a\u667a\u80fd\u52a9\u624b\uff0c\u5f88\u9ad8\u5174\u4e3a\u60a8\u670d\u52a1\uff01\u6211\u5df2\u7ecf\u83b7\u53d6\u4e86\u60a8\u7684\u504f\u597d\u8bbe\u7f6e{r_1}\uff0c\u5e76\u4e3a\u60a8\u8ba1\u7b97\u4e863344545\u52a03444\u7684\u7ed3\u679c\uff0c\u7ed3\u679c\u662f:{c_1}\u3002\u6211\u53ef\u4ee5\u5e2e\u52a9\u60a8\u8fdb\u884c\u5404\u79cd\u4efb\u52a1\uff0c\u5305\u62ec\u4fe1\u606f\u67e5\u8be2\u3001\u8ba1\u7b97\u3001\u5206\u6790\u7b49\u3002\u6709\u4ec0\u4e48\u6211\u53ef\u4ee5\u5e2e\u52a9\u60a8\u7684\u5417\uff1f"
}

测试结果: 您好！我是一个智能助手，很高兴为您服务！我已经获取了您的偏好设置{
  "favorite_city": "Shanghai",
  "temp_unit": "°F"
}，并为您计算了3344545加3444的结果，结果是:3347989。我可以帮助您进行各种任务，包括信息查询、计算、分析等。有什么我可以帮助您的吗？

注意最好不用本地的ollama3:8b模型不够智能，连占位符这种放置逻辑都能出错，我让他自定义name，使用占位符，他永远都是给的相同名字的占位符，得不到你要的结果 ，可以用智普的glm-4.5免费key，基本100%正确，毕竟在线的不可能是这种8b的小模型。