Agent实战03-MCP Server
uv是由 Astral 公司（也是开发了超快 Python 格式化工具Ruff的公司）开发的新工具。它的核心目标是解决 Python 包管理和项目工作流中的速度与复杂度问题。包安装器：替代pip和pip-tools，安装包的速度极快。虚拟环境管理：替代virtualenv或venv，可以快速创建和管理虚拟环境。依赖解析器：替代，快速生成锁定的依赖文件。项目初始化：快速创建新项目。
·
概述
当你理解了 Agent 如何调用 Tools 后，MCP 的概念就会变得非常直观。
这是一个从零基础到实战的 MCP 完整教程。
MCP 是什么？为什么需要它？
在你之前的代码中，Tools 是和你的 Agent 代码紧密耦合的——它们都写在同一个 Python 文件中。MCP (Model Context Protocol) 要解决的核心问题就是：将 Tools 从 Agent 中解耦出来。
一个简单的比喻
- 传统方式：就像在一个厨房里，厨师（Agent）和所有厨具（Tools）都在同一个房间。厨师直接伸手就能拿到刀和锅。
- MCP 方式：厨师（Agent）在厨房，但厨具（Tools）在另一个专业的工具房（MCP Server）里。厨师需要通过一个标准化的传送带（MCP 协议）来获取和使用工具。
这样做的巨大优势：
- 安全性：Tools 在独立的进程中运行，一个工具的崩溃不会影响整个 Agent。
- 语言无关性：Tools 可以用任何语言编写（Python, JavaScript, Go, Rust…），只要遵循 MCP 协议。
- 可复用性：一套 Tools 可以被多个不同的 Agent 使用。
- 易于管理：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能力介绍:
- Resource :是 MCP 中最基本的概念，代表一个具体的、可读的数据实体。你可以把它想象成一个文件、一个数据库查询结果、一个 API 响应或任何一段可供 AI 读取和引用的信息,比如例子中的@mcp.resource(“greeting://default”)没有任何参数就是一个资源。
- ResourceTemplate ：是一个用于发现和生成 Resource 的蓝图或模式。它本身不是资源，而是定义了如何找到或创建一类资源的规则，就是通过参数获取一个资源，比如例子中的@mcp.resource(“greeting://{name}”)。
- Prompt 是 MCPServer 向 AI 客户端注册的预定义提示词模板。它允许服务器为用户提供一些“开箱即用”的、功能强大的对话起点，而不是让用户每次都从零开始编写复杂的提示词，Prompt 是传入一组参数（可能包括关键字），然后返回一个填充好的、完整的、可直接使用的提示词，比如代码中的@mcp.prompt() greet_user 。
- Tools（也称为 Function Calling）允许 AI 执行操作而不仅仅是读取数据。它赋予了 AI 与外部世界交互和写的能力。
MCP Inspector
MCP Inspector 是一个官方开发的、用于可视化调试和监控 MCP（Model Context Protocol）连接和通信的图形化工具。
您可以把它理解为一个 MCP 的 “开发者调试面板” 或 “网络监视器”。
为什么需要它？它能解决什么问题？
在没有 Inspector 之前，调试 MCP Server 非常困难，你只能通过查看终端里打印的日志来猜测 JSON-RPC 消息的往来，效率极低。
MCP Inspector 通过可视化界面完美解决了这个问题：
- 服务器连接与管理
让你可以轻松地添加、配置、启动和停止一个或多个 MCP Server。你不需要手动写配置文件和通过命令行启动，只需在界面点击即可。
- 实时消息流监控
这是最核心的功能。Inspector 会实时显示 Client 和 Server 之间所有的 JSON-RPC 请求和响应，像一个“抓包工具”。
- 你可以看到每条消息的详细内容（方法名、参数、结果、错误）。
- 消息通常用不同颜色标记（如请求、成功响应、错误）。
- 这让你一眼就能看出通信是否成功，参数是否正确，结果是否符合预期。
- 可视化浏览 Server 能力
Inspector 会自动列出已连接 Server 提供的所有能力（Capabilities）：
- Resources & Templates：以文件树等友好形式展示所有可用的资源和模板，你可以直接点击查看内容。
- Prompts：列出所有注册的提示词模板，你可以直接填写参数并测试运行，看会生成什么样的最终提示词。
- Tools：列出所有可用的工具，你可以直接填写参数并调用它们，实时查看执行结果和返回内容。
- 错误诊断
当通信出现错误（例如 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会帮我们自动生成。
本示例包含两部分：
- MCP 服务端：提供「加法工具」「天气查询工具」和「用户偏好资源模板」；
- 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的小模型。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
11
0- 0
已为社区贡献16条内容
所有评论(0)