七台河市网站建设_网站建设公司_Oracle_seo优化
2025/12/20 18:45:15 网站建设 项目流程

一、你遇到的主要问题 & 对应解答

问题你的困惑正确理解 / 解决方法
1. 客户端为什么用 session.get_prompt("greet_user", ...) 而不是 session.greet_user(...)觉得函数名应该直接可用✅ MCP 是动态 RPC 协议,客户端不知道服务器有哪些函数,必须通过名字字符串 + 标准接口调用(如 get_prompt),实现解耦与通用性。
2. session 到底是客户端还是服务器?混淆角色✅ session = ClientSession(...) 是客户端对象,代表与 MCP 服务器的连接。所有 .list_tools().call_tool() 都是客户端发起请求
3. 为什么先 list_tools() 又能直接 call_tool("add")疑惑是否冗余✅ list_* 是可选的发现阶段(用于 UI/自动补全),call_tool 是执行阶段。如果你已知工具名,可跳过 list 直接调用。
4. r.uri 和 resources.resources 为什么嵌套?觉得结构奇怪✅ MCP 响应是标准化的包裹对象(如 ListResourcesResult),.resources 是列表字段,每个元素有 .uri。这是为了协议扩展性(未来加 metadata、权限等)。
5. read_resource 是哪来的?不知道来源✅ 它是 MCP 客户端 SDK 内置的标准方法,对应协议中的 mcp/readResource 方法,用于读取 URI 资源(如 greeting://Alice)。
6. isinstance 的作用是什么?不理解为何需要类型判断✅ 因为 MCP 内容可能是文本、图片、结构化数据等,isinstance(result, TextContent) 可安全访问属性,避免 AttributeError
7. result.structuredContent 是 JSON 吗?混淆序列化概念✅ 它是已解析的 Python dict/list,不是 JSON 字符串。服务器返回结构化数据,客户端 SDK 自动反序列化。
8. handle_sampling_message 中 context 没用到,为何存在?觉得多余✅ context 是预留的请求上下文(含 client session、request ID、认证信息等),用于日志、权限、追踪等扩展。即使当前不用,签名也必须保留以符合框架要求。
9. list_resources() 返回空列表正常吗?怀疑资源没注册成功✅ 完全正常! 动态 URI 模板(如 greeting://{name})通常不列入静态资源清单,但可通过 read_resource("greeting://xxx") 直接访问。

二、关键解决方法汇总

场景推荐做法
调用 Prompt/Tool/Resource使用标准方法:
get_prompt(name, args)
call_tool(name, args)
read_resource(uri)
处理不同类型的内容用 isinstance 安全判断:
python<br>if isinstance(c, TextContent): ...<br>elif isinstance(c, StructuredContent): ...<br>
调试服务器行为查看日志中的 Processing request of type XXXRequest,确认请求类型是否匹配
修复拼写/逻辑错误检查服务器函数返回值(如 "somenoe" → "someone"
扩展 AI 能力实现 handle_sampling_message 并接入真实 LLM(如 OpenAI、Ollama)

三、MCP 核心原理与理论知识

1. MCP 是什么?

  • 全称:Model Context Protocol(模型上下文协议)
  • 目标:为 AI Agent(大模型) 提供标准化方式,访问外部工具(Tools)、数据(Resources)、提示模板(Prompts)
  • 类比:AI 世界的 HTTP + RESTful API

2. 架构模型:客户端-服务器(C/S)

[AI 应用 / IDE / Cherry Studio]   ← stdio / HTTP →   [MCP Server]↑(ClientSession)                             ↑(FastMCP)└─ 发送 JSON-RPC 请求                          └─ 执行 @mcp.tool() 等

3. 三大核心能力

能力用途服务器装饰器客户端方法
Tool执行操作(如计算、API 调用)@mcp.tool()call_tool(name, args)
Resource读取数据(如文件、数据库)@mcp.resource(uri_pattern)read_resource(uri)
Prompt获取预定义提示模板@mcp.prompt()get_prompt(name, args)

4. 通信协议

  • 基于 JSON-RPC 2.0
  • 支持传输方式:stdio(本地子进程)、http(远程服务)
  • 示例请求:
    {"method": "mcp/callTool","params": { "name": "add", "arguments": {"a": 1, "b": 2} }
    }

5. 设计哲学

  • 解耦:AI 应用 ≠ 工具实现,可独立演进
  • 动态发现:运行时获取能力列表,非硬编码
  • 语言无关:只要符合 MCP 协议,Java/Go/Rust 都能做服务器
  • 标准化:统一接口,避免“每个工具一套 API”

四、你的学习成果总结

✅ 你已经掌握了:

  • 如何用 FastMCP 快速搭建 MCP 服务器
  • 如何用 ClientSession 编写客户端调用
  • 理解了 Prompt/Resource/Tool 的区别与用途
  • 能解读日志并验证功能正确性
  • 明白了 MCP 的“动态 RPC”本质

下一步建议:

  1. 尝试让 handle_sampling_message 调用真实 LLM
  2. 用 Cherry Studio 连接你的服务器做可视化测试
  3. 实现一个“读取本地文件”的 Resource
  4. 探索 MCP 的权限、日志、错误处理机制
# server 构建mcp服务器
# 导入 FastMCP 类,用于创建符合 MCP协议的服务器
from mcp.server.fastmcp import FastMCP
# 创建一个名为"Demo"的MCP服务器实例
# FastMCP 会自动将工具(tool),资源(resource),提示词(prompt)注册为标准的MCP接口
mcp = FastMCP("Demo")
# 定义一个工具(TOOL)
# 使用@mcp.too()装饰器将函数注册为MCP工具
# 工具是客户端可以远程调用的函数
@mcp.tool()
def add(a:int,b:int) -> int:"""将两个数字相加"""return a+b
# 定义一个资源(Resource)
# 使用@mcp.resource()装饰器定义一个URI模式为"greeting://{name}"的资源
# 当客户端访问该URI时,会触发次函数并传入name参数
@mcp.resource("greeting://{name}")
def get_greeting(name:str)->str:"""生成一个问候语"""# 根据传入的名字name返回对应的问候语return f"Hello, {name}"
# 定义一个提示词(Prompt)
# 使用@mcp.prompt()装饰器注册一个提示词函数
@mcp.prompt()
def greet_user(name:str,style:str="friendly")->str:"""生成一个问候用户的提示词"""styles = {"friendly":"please write a warm,friendly greeting","formal":"please write a formal,professional greeting","casual":"please write a casual,relaxed greeting",}# 根据style 参数选择合适的问候语模式,如果style不在字典中默认使用friendly风格return f"{styles.get(style,styles['friendly'])} for somenoe named {name}"
if __name__=="__main__":# 启动mcp服务器,使用stdio(标准输入输出)方式进行通信# 这种方式适用于本地子进程模式,客户端通过stdio_client连接mcp.run("stdio")
# client
# 导入必要的模块
import asyncio  # 用于异步编程支持
import os
# 用于操作系统相关功能,如环境变量访问
from pydantic import AnyUrl  # 用于URL类型验证
# 导入MCP相关模块
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client  # 用于创建stdio客户端连接
from mcp.shared.context import RequestContext  # 用于处理请求上下文
# =====================
# 服务器参数配置
# =====================
# 配置如何启动MCP服务器的参数
server_params = StdioServerParameters(command="uv",  # 使用uv命令来运行服务器args=["run", "server.py"],  # 运行server.py文件的参数env={"UV_INDEX": os.environ.get("UV_INDEX", "")},  # 设置环境变量
)
# =====================
# 消息处理回调函数
# =====================
# 定义处理采样消息的回调函数,当模型需要生成内容时会被调用
async def handle_sampling_message(context: RequestContext[ClientSession, None],params: types.CreateMessageRequestParams
) -> types.CreateMessageResult:# 打印接收到的采样请求消息print(f"Sampling request: {params.messages}")# 返回一个预定义的响应消息return types.CreateMessageResult(role="assistant",  # 角色为助手content=types.TextContent(  # 消息内容type="text",  # 内容类型为文本text="Hello, world! from model",  # 实际文本内容),model="gpt-3.5-turbo",  # 使用的模型名称stopReason="endTurn",# 停止原因)
# =====================
# 主运行函数
# =====================
async def run():try:# 使用stdio_client创建与服务器的连接,通过上下文管理器确保正确关闭async with stdio_client(server_params) as (read, write):# 创建客户端会话,传入读写流和采样回调函数async with ClientSession(read, write,sampling_callback=handle_sampling_message) as session:# 初始化客户端会话await session.initialize() # 握手# =====================# 列出可用的提示词# =====================# 获取服务器上所有可用的提示词列表prompts = await session.list_prompts()print(f"Available prompts: {[p.name for p in prompts.prompts]}")# 如果存在提示词,则获取并使用其中一个if prompts.prompts:# 获取名为"greet_user"的提示词,传入参数name="Alice"和style = "friendly"prompt = await session.get_prompt("greet_user",arguments={"name": "Alice", "style": "friendly"})# 打印获取到的提示词内容print(f"Prompt result: {prompt.messages[0].content}")# =====================# 列出可用资源# =====================# 获取服务器上所有可用资源的列表resources = await session.list_resources()print(f"Available resources: {[r.uri for r in resources.resources]}")# =====================# 列出可用工具# =====================# 获取服务器上所有可用工具的列表tools = await session.list_tools()print(f"Available tools: {[t.name for t in tools.tools]}")# =====================# 读取资源内容# =====================# 读取URI为"greeting://World"的资源内容resource_content = await session.read_resource(AnyUrl("greeting://World"))content_block = resource_content.contents[0]# 检查内容块是否为文本类型,如果是则打印内容if isinstance(content_block, types.TextResourceContents):print(f"Resource content: {content_block.text}")# =====================# 调用工具# =====================# 调用名为"add"的工具,传入参数a=5和b=3result = await session.call_tool("add", arguments={"a": 5, "b": 3})# 处理工具返回的非结构化内容result_unstructured = result.content[0]if isinstance(result_unstructured, types.TextContent):print(f"Tool result: {result_unstructured.text}")# 处理工具返回的结构化内容result_structured = result.structuredContentprint(f"Structured tool result: {result_structured}")except Exception as e:# 错误处理:打印连接错误信息print(f"Error connecting to server: {e}")print("请确保服务器文件存在,并能通过 'uv run server-stdio.py' 正常运行")
# =====================
# 程序入口点
# =====================
def main():# 运行异步函数run()asyncio.run(run())
# 程序主入口
if __name__ == '__main__':main()

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询