1. 工具列表 #

• 什么是工具？

如何基于 JSON-RPC 协议扩展，实现“工具列表”能力。

通过在 FastMCP 服务器上注册工具（如用 `@mcp.tool()装饰器），服务器会自动收集所有注册的工具，并在收到tools/list` 方法调用时，返回工具名称、描述和参数 schema 等信息。

客户端调用 session.list_tools() 即可获得所有可用工具的信息。例如：

tools = session.list_tools()
print([t.name for t in tools.tools])

这样，客户端无需关心工具的具体实现细节，即可枚举、展示和后续调用服务端的各类工具，有利于灵活集成自动化和扩展。

npx @modelcontextprotocol/inspector uv --directory D:/forever/docs/mcpsdk2 run 1.mcp.py serve

1.1. 1.mcp.py #

1.mcp.py

# 导入 sys 模块
import sys
# 导入 mcp_lite 模块
from mcp_lite import ClientSession,StdioServerParameters
# 导入 mcp_lite.client.stdio 模块
from mcp_lite.client.stdio import stdio_client
# 导入 mcp_lite.server.fastmcp 模块
from mcp_lite.server.fastmcp import FastMCP
# 创建一个名为 "Hello-MCP" 的 FastMCP 服务器对象
mcp = FastMCP(name="Hello-MCP")
# 通过装饰器注册工具，工具名称自动取函数名 hello
+@mcp.tool()
+def hello():
    # 返回字符串作为工具实现
+   return "Hello from MCP server!"
# 定义客户端启动函数
def run_client():
    # 构造 StdioServerParameters，用 python 启动本文件并传入"serve"参数
    server_params = StdioServerParameters(command="python", args=[__file__, "serve"])
     # 使用上下文管理器启动 stdio_client，返回读写流
    with stdio_client(server_params) as (read, write):
        # 创建客户端会话对象，绑定读写流
        session = ClientSession(read, write)
         # 初始化会话，完成握手
        session.initialize()
        # 获取服务器端可用工具列表
+       tools = session.list_tools()
        # 打印所有工具名称
+       print("[Tools]", [t.name for t in tools.tools])

# 定义以 stdio 运行的服务器端函数
def run_server_stdio():
    # 在标准错误流打印启动提示
    print("启动 MCP 服务器（stdio 模式）...", file=sys.stderr)
    # 以 stdio 方式运行 mcp 服务器
    mcp.run(transport="stdio")

# 主入口函数，根据命令行参数决定运行模式
def main():
    # 如果参数含有"serve"，以服务器模式运行
    if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        run_server_stdio()
    else:
        # 否则启动客户端测试流程，自动拉起服务端
        print("启动 MCP 客户端测试...")
        print("将自动启动服务器进程并测试连接")
        run_client()

# 判断是否直接运行此文件
if __name__ == "__main__":
    main()

官方代码

# 导入 sys 模块
import sys

# 导入 mcp 的 ClientSession 和 StdioServerParameters
from mcp import ClientSession,StdioServerParameters

# 导入 mcp.client.stdio 中的 stdio_client
from mcp.client.stdio import stdio_client

# 导入 mcp.server.fastmcp 中的 FastMCP
from mcp.server.fastmcp import FastMCP

# 导入 asyncio 模块，处理异步任务
import asyncio

# 创建一个名为 "Hello-MCP" 的 FastMCP 服务器对象
mcp = FastMCP(name="Hello-MCP")

# 使用装饰器注册工具，工具名为 hello
+@mcp.tool()
+def hello():
+    # 返回字符串作为工具实现
+    return "Hello from MCP server!"

# 定义异步客户端启动函数
async def run_client():
    # 构造服务器参数对象，命令为 python，参数为当前文件和 'serve'
    server_params = StdioServerParameters(command="python", args=[__file__, "serve"])
    # 使用异步上下文管理器启动 stdio_client，获得读写流
    async with stdio_client(server_params) as (read, write):
+       # 用异步上下文管理器创建会话，并绑定读写流
+       async with ClientSession(read, write) as session:
            # 初始化会话，完成握手过程
            await session.initialize()
+           # 获取服务器端的工具列表
+           tools = await session.list_tools()
+           # 打印所有工具的名称
+           print("[Tools]", [t.name for t in tools.tools])

# 定义以 stdio 模式运行服务器的函数
def run_server_stdio():
    # 在标准错误流中打印服务器启动提示
    print("启动 MCP 服务器（stdio 模式）...", file=sys.stderr)
    # 以 stdio 方式运行 MCP 服务器
    mcp.run(transport="stdio")

# 主入口函数，根据参数判断是服务器还是客户端模式
def main():
    # 如果命令行参数为 'serve'，则以服务器模式运行
    if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        run_server_stdio()
    else:
        # 否则作为客户端测试，自动启动服务器进程
        print("启动 MCP 客户端测试...")
        print("将自动启动服务器进程并测试连接")
        asyncio.run(run_client())

# 判断是否直接运行本文件
if __name__ == "__main__":
    main()

1.2. session.py #

mcp_lite/client/session.py

import sys
# 导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.types 模块导入相关类型和常量
from mcp_lite.types import (
      InitializeRequestParams,    # 导入初始化请求参数结构体
      InitializeRequest,          # 导入初始化请求结构体
      LATEST_PROTOCOL_VERSION,    # 导入协议最新版本号
      ClientCapabilities,         # 导入客户端能力描述
      Implementation,             # 导入实现信息结构体
      InitializedNotification,    # 导入初始化完成通知
      InitializeResult,           # 导入初始化请求返回结构体
      JSONRPCRequest,             # 导入 JSONRPC 请求类型
      JSONRPCResponse,            # 导入 JSONRPC 响应类型
      JSONRPCError,               # 导入 JSONRPC 错误类型
      JSONRPCNotification,        # 导入 JSONRPC 通知类型
+     ListToolsRequest,           # 导入获取工具列表请求
+     ListToolsResult,            # 导入获取工具列表响应结构体
)
# 定义 MCPError 异常类，用于表示协议相关错误
class MCPError(Exception):
    # 初始化方法，接收错误码、错误信息和可选的数据
    def __init__(self, code, message, data=None):
        # 将错误码、错误信息、附加数据赋值为实例属性
        self.code, self.message, self.data = code, message, data
        # 调用父类 Exception 的初始化，只传递错误信息
        super().__init__(message)

    # 类方法，利用 JSONRPCError 对象创建 MCPError 实例
    @classmethod
    def from_jsonrpc(cls, err):
        # 从 JSONRPCError.error 对象中取出 code/message/data 创建 MCPError
        return cls(err.error.code, err.error.message, err.error.data)
# 定义 ClientSession 类，用于描述客户端会话
class ClientSession:
    # 构造方法，接收读取流和写入流
    def __init__(self, read_stream, write_stream):
        # 赋值读取流和写入流
        self._read, self._write = read_stream, write_stream
        # 初始化请求 id 计数器为 0
        self._req_id = 0
    # 内部方法，发送请求并同步等待响应
    def _request(self, req):
        # 当前请求 id
        rid = self._req_id
        # 自增请求 id，确保下次唯一
        self._req_id += 1
        # 对请求对象进行序列化，转为 dict
        d = req.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 构建 JSONRPCRequest，再封装入 SessionMessage，通过写入流发送
        jreq = JSONRPCRequest(jsonrpc="2.0", id=rid, method=d["method"], params=d.get("params"))
        # 打印请求信息到标准错误流
        print("[Client] Request:", jreq.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
        # 将请求对象封装成 SessionMessage 后发送
        self._write.send(SessionMessage(message=jreq))
        # 循环等待响应
        while True:
            # 从读取流取出一条消息
            msg = self._read.get()
            # 若取到 None，说明连接断开，抛出异常
            if msg is None:
                raise MCPError(-32000, "Connection closed")
            # 若消息类型不是 SessionMessage，忽略继续等
            if not isinstance(msg, SessionMessage):
                continue
            # 取消息内容
            m = msg.message
            # 若为 Response 或 Error，且 id 匹配
            if isinstance(m, (JSONRPCResponse, JSONRPCError)) and getattr(m, "id", None) == rid:
                print("[Client] Response:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
                # 若为错误，抛出 MCPError 异常
                if isinstance(m, JSONRPCError):
                    raise MCPError.from_jsonrpc(m)
                # 为正常响应则返回结果数据
                return m.result
    # 内部方法，发送通知消息
    def _notify(self, n):
        # 通知对象序列化为 dict
        d = n.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 封装为 JSONRPCNotification，再封装成 SessionMessage 后发出
        self._write.send(SessionMessage(
            message=JSONRPCNotification(
                jsonrpc="2.0",
                method=d["method"],
                params=d.get("params"),
            )
        ))            
    # 初始化过程，发送初始化请求，收到响应后再发初始化完成通知
    def initialize(self):
        # 构造初始化请求，并发送等待返回
        r = self._request(InitializeRequest(
            params=InitializeRequestParams(
                protocol_version=LATEST_PROTOCOL_VERSION,         # 协议版本号
                capabilities=ClientCapabilities(),                # 客户端能力
                client_info=Implementation(name="mcp_lite", version="0.1.0"),  # 客户端信息
            )
        ))
        # 发送"初始化完成"通知
        self._notify(InitializedNotification())
        # 用 InitializeResult 验证响应并返回
        return InitializeResult.model_validate(r, by_name=False)
        # 获取工具列表
+   def list_tools(self):
        # 发送 ListToolsRequest 请求，使用 ListToolsResult 校验并返回
+       return ListToolsResult.model_validate(self._request(ListToolsRequest()), by_name=False)    

1.3. fastmcp.py #

mcp_lite/server/fastmcp.py

# 导入 sys 模块，用于标准输入输出操作
import sys
# 导入 inspect 用于函数签名分析
+import inspect
# 导入 asyncio 用于异步操作
+import asyncio
# 从 mcp_lite.message 模块导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.server 模块导入 stdio 子模块
from mcp_lite.server import stdio
# 从 mcp_lite.types 模块导入所有相关类型和类
from mcp_lite.types import (
     JSONRPCRequest,            # JSONRPC 请求类型
     JSONRPCError,              # JSONRPC 错误类型
     ErrorData,                 # 错误数据类型
     InitializeResult,          # 初始化响应类型
     LATEST_PROTOCOL_VERSION,   # 最新协议版本常量
     ToolsCapability,           # 工具能力类型
     ServerCapabilities,        # 服务器功能特性类型
     Implementation,            # 实现信息类型
     JSONRPCResponse,           # JSONRPC 响应类型
+    ListToolsResult,           # 工具列表响应类型
+    Tool,                      # 工具类型
)
# 内部函数：根据函数定义提取参数信息并生成 schema
+def _schema(fn):
    # 获取函数签名
+   sig = inspect.signature(fn)
    # 用于存储参数属性的字典
+   props = {}
    # 用于存储必需参数名的列表
+   req = []
    # 遍历所有参数
+   for n, p in sig.parameters.items():
        # 跳过以 "_" 开头的参数
+       if n.startswith("_"):
+           continue
        # 所有参数类型都设为 string，title 为参数名
+       props[n] = {"type": "string", "title": n}
        # 如果参数没有默认值，则为必需参数
+       if p.default is inspect.Parameter.empty:
+           req.append(n)
    # 返回对象类型 schema，包括属性和必需项目
+   return {"type": "object", "properties": props, "required": req}
# 内部类：用于封装工具函数
+class _Tool:
    # 初始化方法，接收目标函数、工具名和描述
+   def __init__(self, fn, name=None, desc=None):
        # 保存原始函数引用
+       self.fn = fn
        # 工具名称，优先采用传参，否则用函数名
+       self.name = name or fn.__name__
        # 工具描述，优先采用传参，否则用函数 docstring
+       self.desc = desc or (fn.__doc__ or "").strip()
        # 自动生成入参 schema
+       self.schema = _schema(fn)
        # 判断函数是否为异步（协程）函数
+       self.async_fn = asyncio.iscoroutinefunction(fn)

    # 转换为 Tool 类型的对象
+   def to_tool(self):
        # 返回 Tool 类型实例
+       return Tool(name=self.name, description=self.desc or None, input_schema=self.schema)

    # 执行工具函数，参数为字典类型
+   def run(self, args):
        # 如果是异步函数则使用 asyncio.run 执行，否则同步调用
+       if self.async_fn:
+           return asyncio.run(self.fn(**args))
+       return self.fn(**args)
# 主服务类 FastMCP
class FastMCP:
    # 初始化方法，支持自定义服务器名称
    def __init__(self, name="mcp-server"):
        # 保存服务器名称
        self.name = name
        # 初始化工具池为字典
+       self._tools = {}
    # 工具注册装饰器，支持自定义工具名和描述
+   def tool(self, name=None, description=None):
        # 装饰器实际函数
+       def deco(fn):
            # 创建 _Tool 实例
+           t = _Tool(fn, name, description)
            # 注册进工具池
+           self._tools[t.name] = t
            # 返回原始函数
+           return fn
        # 返回装饰器
+       return deco    
     # 实际 JSONRPC 方法分发与业务处理
    def _handle(self, req):
        # 拆分 method、params、id
        method, params, rid = req.method, req.params or {}, req.id

        # 处理初始化请求
        if method == "initialize":
            # 组装初始化响应，包括协议版本、能力与服务信息
            r = InitializeResult(
                protocol_version=LATEST_PROTOCOL_VERSION,
                capabilities=ServerCapabilities(tools=ToolsCapability()),
                server_info=Implementation(name=self.name, version="0.1.0"),
            )
            # 返回 JSONRPCResponse 包装的结果
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
        # 处理工具列表请求
+       if method == "tools/list":
            # 构建工具列表结果
+           r = ListToolsResult(tools=[t.to_tool() for t in self._tools.values()])
            # 返回 JSONRPCResponse 包装的结果
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))    
        # 不支持的方法，返回 "方法未找到" 错误（code -32601）
        return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32601, message=f"Method not found: {method}"))    
    # 处理 SessionMessage 消息，主要路由 JSONRPCRequest
    def _handle_msg(self, msg):
        # 判断消息类型不是 SessionMessage 则忽略
        if not isinstance(msg, SessionMessage):
            return None
        # 获取消息体
        m = msg.message
        # 只处理 JSONRPCRequest 类型
        if not isinstance(m, JSONRPCRequest):
            return None
        print("[Server] Request:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)    
        try:
            # 调用实际业务处理方法
            resp = self._handle(m)
            print("[Server] Response:", resp.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return resp
        except Exception as e:
            # 捕获异常，返回标准 JSONRPCError（code -32603，内部错误）
            err = JSONRPCError(jsonrpc="2.0", id=m.id, error=ErrorData(code=-32603, message=str(e)))
            print("[Server] Response (error):", err.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return err
    # 运行服务方法，默认采用 stdio 传输方式
    def run(self, transport="stdio"):
        # 仅支持 stdio，其他方式抛出异常
        if transport != "stdio":
            raise ValueError(f"unsupported transport: {transport}")
        # 启动 stdio server，当前对象 _handle_msg 作为消息处理回调
        stdio.stdio_server(self._handle_msg)    

1.4. types.py #

mcp_lite/types.py

# 导入 Any 和 Literal 类型注解
from typing import Any, Literal
# 从 pydantic 导入 BaseModel、ConfigDict、Field、TypeAdapter
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
# 导入 pydantic 的 to_camel 驼峰命名生成器
from pydantic.alias_generators import to_camel
# 定义 RequestId 类型，既可以是 int 也可以是 str
RequestId = int | str
# 定义 MCP 的基础模型类，支持驼峰命名和按名称填充
class MCPModel(BaseModel):
    # 指定 Pydantic 模型配置：使用驼峰形式命名和按名称填充
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
# 定义客户端能力结构体
class ClientCapabilities(MCPModel):
    # 可选字段：客户端实验性能力扩展，键为 str，值为字典
    experimental: dict[str, dict[str, Any]] | None = None
# 定义服务器或客户端的实现信息结构体
class Implementation(MCPModel):
    # 实现的名称
    name: str = ""
    # 实现的版本号
    version: str = ""
    # 可选字段：人类可读的标题
    title: str | None = None
    # 可选字段：实现的描述
    description: str | None = None    
# 定义初始化请求参数结构体
class InitializeRequestParams(MCPModel):
    # 协议版本号
    protocol_version: str = ""
    # 可选字段：客户端能力描述
    capabilities: ClientCapabilities = None
    # 可选字段：客户端实现信息
    client_info: Implementation = None
# 定义初始化请求结构体
class InitializeRequest(MCPModel):
    # 方法名，固定为 "initialize"
    method: Literal["initialize"] = "initialize"
    # 可选字段：参数，类型为 InitializeRequestParams
    params: InitializeRequestParams = None    
# 当前协议的最新版本号
LATEST_PROTOCOL_VERSION = "2024-11-05"
# 定义工具相关能力结构体
class ToolsCapability(MCPModel):
    # 工具列表是否发生变化，可选布尔类型
    list_changed: bool | None = None
# 定义服务端能力描述结构体
class ServerCapabilities(MCPModel):
    # 可选字段：实验性能力扩展
    experimental: dict[str, dict[str, Any]] | None = None
    # 可选字段：工具能力
    tools: ToolsCapability | None = None
# 定义初始化响应结构体
class InitializeResult(MCPModel):
    # 协议版本号
    protocol_version: str = ""
    # 可选字段：服务端能力描述
    capabilities: ServerCapabilities = None
    # 可选字段：服务端实现信息
    server_info: Implementation = None
    # 可选字段：初始化说明信息
    instructions: str | None = None
# 定义客户端初始化完成通知结构体
class InitializedNotification(MCPModel):
    # 方法名，固定为 "notifications/initialized"
    method: Literal["notifications/initialized"] = "notifications/initialized"
    # 可选字段：通知参数，可以为字典或 None
    params: dict[str, Any] | None = None    
# 定义 JSONRPC 请求的数据结构
class JSONRPCRequest(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 请求 ID，可以为 int 或 str 类型
    id: RequestId = None
    # 方法名称，字符串类型
    method: str = ""
    # 方法参数，为一个字典或 None
    params: dict[str, Any] | None = None    

# 定义 JSONRPC 响应的数据结构
class JSONRPCResponse(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 响应的 ID，需要与请求 ID 匹配
    id: RequestId = None
    # 响应结果，可以为字典或 None
    result: dict[str, Any] = None   
# 定义错误的数据结构
class ErrorData(BaseModel):
    # 错误码，默认值为 0
    code: int = 0
    # 错误信息，默认值为空字符串
    message: str = ""
    # 附加的错误数据，可以为任意类型或 None
    data: Any = None     
# 定义 JSONRPC 错误消息的数据结构
class JSONRPCError(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 错误对应的请求 ID，可以为 None
    id: RequestId | None = None
    # 错误的详细信息，类型为 ErrorData
    error: ErrorData = None    
# 定义 JSONRPC 通知的数据结构（没有 id 字段）
class JSONRPCNotification(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 通知的方法名称
    method: str = ""
    # 通知参数，可以为字典或者 None
    params: dict[str, Any] | None = None    

# 定义所有 JSONRPC 消息的联合类型
JSONRPCMessage = JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError

# 定义 JSONRPC 消息适配器，用于类型自动推断和校验
jsonrpc_message_adapter = TypeAdapter(JSONRPCMessage)
# 定义工具描述数据结构体
+class Tool(MCPModel):
    # 工具名称
+   name: str = ""
    # 可选字段：工具描述
+   description: str | None = None
    # 工具输入参数的 schema，默认为空字典
+   input_schema: dict[str, Any] = Field(default_factory=dict)
    # 可选字段：工具输出 schema
+   output_schema: dict[str, Any] | None = None
# 定义获取工具列表请求结构体
+class ListToolsRequest(MCPModel):
    # 方法名，固定为 "tools/list"
+   method: Literal["tools/list"] = "tools/list"
    # 可选字段：参数，可以为字典或 None
+   params: dict[str, Any] | None = None

# 定义获取工具列表响应结构体
+class ListToolsResult(MCPModel):
    # 工具列表，默认为空列表
+   tools: list[Tool] = []
    # 可选字段：分页游标，可为 None
+   next_cursor: str | None = None

# 定义获取工具列表响应结构体
+class ListToolsResult(MCPModel):
    # 工具列表，默认为空列表
+   tools: list[Tool] = []
    # 可选字段：分页游标，可为 None
+   next_cursor: str | None = None    

1.5 工作流程 #

1.5.1 整体功能 #

在初始化之后，增加 MCP 工具（Tools） 能力：

1. 服务端：用 `@mcp.tool()` 注册工具函数
2. 客户端：通过 session.list_tools() 获取工具列表
3. 协议：新增 tools/list 方法，返回工具名称、描述、参数 schema

1.5.2 1.mcp.py注册工具并调用 list_tools #

+@mcp.tool()
+def hello():
+   return "Hello from MCP server!"

• `@mcp.tool()：把hello` 注册为 MCP 工具
• 工具名默认用函数名 hello
• 返回值会作为工具执行结果

        session.initialize()
+       tools = session.list_tools()
+       print("[Tools]", [t.name for t in tools.tools])

• 初始化完成后调用 session.list_tools()
• 打印所有工具名称，例如 [Tools] ['hello']

1.5.3 1.2 session.py新增 list_tools #

+     ListToolsRequest,           # 导入获取工具列表请求
+     ListToolsResult,            # 导入获取工具列表响应结构体

+   def list_tools(self):
+       return ListToolsResult.model_validate(self._request(ListToolsRequest()), by_name=False)

• list_tools()：发送 ListToolsRequest（method 为 tools/list）
• 通过 _request 等待 JSON-RPC 响应
• 用 ListToolsResult.model_validate 校验并返回
• 返回对象包含 tools 列表，每个元素是 Tool（name、description、input_schema 等）

1.5.4 1.3 fastmcp.py工具注册与 tools/list 处理 #

1.5.5 _schema：从函数签名生成 input_schema #

# 定义_schema函数，根据函数签名生成JSON Schema
def _schema(fn):
    # 获取函数的参数签名
    sig = inspect.signature(fn)
    # 用于存储参数的属性描述
    props = {}
    # 用于存储必需参数名
    req = []
    # 遍历所有参数
    for n, p in sig.parameters.items():
        # 跳过以下划线开头的参数
        if n.startswith("_"):
            continue
        # 为每个参数生成属性配置（类型为string，标题为参数名）
        props[n] = {"type": "string", "title": n}
        # 如果参数没有默认值，则为必需参数
        if p.default is inspect.Parameter.empty:
            req.append(n)
    # 返回符合JSON Schema规范的input_schema
    return {"type": "object", "properties": props, "required": req}

• 用 inspect.signature 获取函数参数
• 跳过以 _ 开头的参数
• 每个参数生成 {"type": "string", "title": 参数名}
• 无默认值的参数加入 required
• 返回 JSON Schema 格式的 input_schema

1.5.6 _Tool：工具封装类 #

# 工具封装类
class _Tool:
    # 构造函数，接收函数对象fn，可选名称name和描述desc
    def __init__(self, fn, name=None, desc=None):
        # 保存原始函数对象
        self.fn = fn
        # 工具名，优先取参数name，否则用函数名
        self.name = name or fn.__name__
        # 工具描述，优先取desc，否则取docstring的内容并去除首尾空白
        self.desc = desc or (fn.__doc__ or "").strip()
        # 根据函数签名生成输入参数的schema
        self.schema = _schema(fn)
        # 判断该函数是否为异步函数，结果为布尔值
        self.async_fn = asyncio.iscoroutinefunction(fn)

    # 转换为MCP Tool对象，用于列表展示
    def to_tool(self):
        return Tool(name=self.name, description=self.desc or None, input_schema=self.schema)

    # 执行工具函数，args为参数字典，根据是否异步分别处理
    def run(self, args):
        if self.async_fn:
            # 如果是异步函数，使用asyncio.run运行
            return asyncio.run(self.fn(**args))
        # 否则直接同步调用
        return self.fn(**args)

• name：优先用传入的 name，否则用函数名
• desc：优先用传入的 description，否则用 docstring
• schema：由 _schema(fn) 生成
• to_tool()：转成 MCP 的 Tool 类型，用于 tools/list 响应
• run(args)：用字典 args 调用 fn(**args)，支持同步和异步函数

1.5.7 FastMCP.tool() 装饰器 #

# 构造函数，接收服务端名称参数，默认为"mcp-server"
def __init__(self, name="mcp-server"):
    # 保存服务端名称
    self.name = name
    # 初始化工具字典，用于存储已注册的工具
    self._tools = {}

# 工具注册方法，可以通过装饰器方式注册工具，支持自定义名称和描述
def tool(self, name=None, description=None):
    # 装饰器内层函数，接收要注册的函数对象
    def deco(fn):
        # 用_Tool类封装函数与元数据
        t = _Tool(fn, name, description)
        # 按工具名保存到_tools字典中，便于后续快速查找
        self._tools[t.name] = t
        # 返回原始函数对象，不做修改
        return fn
    # 返回装饰器
    return deco

• _tools：字典，key 为工具名，value 为 _Tool 实例
• `@mcp.tool()或@mcp.tool(name="xxx", description="yyy")会把函数包装成_Tool` 并注册

1.5.8 处理 tools/list 请求 #

if method == "tools/list":
    r = ListToolsResult(tools=[t.to_tool() for t in self._tools.values()])
    return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

• 收到 tools/list 时，遍历 _tools，把每个 _Tool 转成 Tool 并放入 ListToolsResult.tools
• 返回 JSON-RPC 响应，result 为 ListToolsResult 的字典形式

1.5.9 1.5 types.py工具相关类型 #

1.5.10 Tool #

# 定义工具描述数据结构体，继承自 MCPModel
class Tool(MCPModel):
    # 工具名称，默认为空字符串
    name: str = ""
    # 工具描述，允许为 None
    description: str | None = None
    # 工具输入参数的 schema，默认为空字典
    input_schema: dict[str, Any] = Field(default_factory=dict)
    # 工具输出 schema，默认为 None
    output_schema: dict[str, Any] | None = None

• 工具描述：名称、描述、输入/输出 schema

1.5.11 ListToolsRequest #

# 定义用于获取工具列表的请求模型，继承自 MCPModel
class ListToolsRequest(MCPModel):
    # 请求方法名称，固定为 "tools/list"
    method: Literal["tools/list"] = "tools/list"
    # 可选的请求参数，类型为字典或 None，通常不需要
    params: dict[str, Any] | None = None

• 请求 method 固定为 tools/list

1.5.12 ListToolsResult #

# 定义获取工具列表响应的数据结构体，继承自 MCPModel
class ListToolsResult(MCPModel):
    # 工具列表，类型为 Tool 对象的列表，默认为空列表
    tools: list[Tool] = []
    # 可选的分页游标，支持分页（当前实现未用），默认为 None
    next_cursor: str | None = None

• tools：工具列表
• next_cursor：分页游标（当前实现未使用）

1.5.13 消息流示意 #

客户端                              服务端
  │                                    │
  │  InitializeRequest                 │
  │ ─────────────────────────────────►│
  │  InitializeResult                  │
  │ ◄─────────────────────────────────│
  │  InitializedNotification           │
  │ ─────────────────────────────────►│
  │                                    │
  │  ListToolsRequest (tools/list)     │
  │ ─────────────────────────────────►│
  │                                    │  _handle("tools/list")
  │                                    │  遍历 _tools，生成 ListToolsResult
  │  ListToolsResult                   │
  │ ◄─────────────────────────────────│
  │                                    │
  │  tools.tools = [Tool(name="hello", ...)]
  │                                    │

1.5.14 小结 #

当前实现只完成了 列出工具，还没有实现 tools/call（调用工具）。要真正执行工具，还需要在服务端增加对 tools/call 的处理，并在客户端增加 session.call_tool() 之类的方法。

2. 调用工具 #

本节新增了对调用工具（tools/call）的介绍，主要包括两方面：

1. 服务端（FastMCP）实现了 tools/call 方法分发。收到该请求后，服务端会根据请求参数中的工具名查找已注册工具（即 _tools 字典），然后自动解析参数并运行对应的 Python 函数，将运行结果封装为 CallToolResult 结构体返回。

2. 客户端则增加了 session.call_tool(name, arguments) 方法。该方法会构造并发送 CallToolRequest，带上目标工具名称和参数字典，收取响应后，以 CallToolResult 校验模型返回。响应中的 content 字段为一个内容块列表，一般包含文本结果，可遍历提取。

这样，客户端和服务端共同完成了工具从注册、列举到实际调用的端到端流程，定义和使用都极为简洁。

1.mcp.py，可以用 session.call_tool("hello", arguments={}) 的方式，实现远程函数的自动调用和结构化输出。

2.1. 1.mcp.py #

1.mcp.py

# 导入 sys 模块
import sys
# 导入 mcp_lite 模块
from mcp_lite import ClientSession,StdioServerParameters
# 导入 mcp_lite.client.stdio 模块
from mcp_lite.client.stdio import stdio_client
# 导入 mcp_lite.server.fastmcp 模块
from mcp_lite.server.fastmcp import FastMCP
# 创建一个名为 "Hello-MCP" 的 FastMCP 服务器对象
mcp = FastMCP(name="Hello-MCP")
# 通过装饰器注册工具，工具名称自动取函数名 hello
@mcp.tool()
def hello():
    # 返回字符串作为工具实现
    return "Hello from MCP server!"
# 定义客户端启动函数
def run_client():
    # 构造 StdioServerParameters，用 python 启动本文件并传入"serve"参数
    server_params = StdioServerParameters(command="python", args=[__file__, "serve"])
     # 使用上下文管理器启动 stdio_client，返回读写流
    with stdio_client(server_params) as (read, write):
        # 创建客户端会话对象，绑定读写流
        session = ClientSession(read, write)
         # 初始化会话，完成握手
        session.initialize()
        # 获取服务器端可用工具列表
        tools = session.list_tools()
        # 打印所有工具名称
        print("[Tools]", [t.name for t in tools.tools])
         # 调用名为 "hello" 的工具，传入空参数
+       result = session.call_tool("hello", arguments={})
        # 用于收集文本返回内容
+       tool_texts = []
        # 遍历每个内容块
+       for block in result.content:
            # 如果内容块是字典且类型为"text"，提取 "text" 字段
+           if isinstance(block, dict) and block.get("type") == "text":
+               tool_texts.append(block.get("text", ""))
            # 如果内容块有 text 属性，直接取出 text
+           elif hasattr(block, "text"):
+               tool_texts.append(block.text)
        # 打印调用 hello 工具的响应内容
+       print("[Call hello]", " | ".join(tool_texts) or str(result.structured_content))

# 定义以 stdio 运行的服务器端函数
def run_server_stdio():
    # 在标准错误流打印启动提示
    print("启动 MCP 服务器（stdio 模式）...", file=sys.stderr)
    # 以 stdio 方式运行 mcp 服务器
    mcp.run(transport="stdio")

# 主入口函数，根据命令行参数决定运行模式
def main():
    # 如果参数含有"serve"，以服务器模式运行
    if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        run_server_stdio()
    else:
        # 否则启动客户端测试流程，自动拉起服务端
        print("启动 MCP 客户端测试...")
        print("将自动启动服务器进程并测试连接")
        run_client()

# 判断是否直接运行此文件
if __name__ == "__main__":
    main()

官方代码

# 导入 sys 模块
import sys
# 导入 mcp_lite 模块
from mcp import ClientSession,StdioServerParameters
# 导入 mcp_lite.client.stdio 模块
from mcp.client.stdio import stdio_client
# 导入 mcp_lite.server.fastmcp 模块
from mcp.server.fastmcp import FastMCP
# 导入 asyncio 模块，处理异步任务
import asyncio
# 创建一个名为 "Hello-MCP" 的 FastMCP 服务器对象
mcp = FastMCP(name="Hello-MCP")
# 通过装饰器注册工具，工具名称自动取函数名 hello
@mcp.tool()
def hello():
    # 返回字符串作为工具实现
    return "Hello from MCP server!"
# 定义客户端启动函数
async def run_client():
    # 构造 StdioServerParameters，用 python 启动本文件并传入"serve"参数
    server_params = StdioServerParameters(command="python", args=[__file__, "serve"])
     # 使用上下文管理器启动 stdio_client，返回读写流
    async with stdio_client(server_params) as (read, write):
        # 用异步上下文管理器创建会话，并绑定读写流
        async with ClientSession(read, write) as session:
            # 初始化会话，完成握手过程
            await session.initialize()
            # 获取服务器端可用工具列表
            tools = await session.list_tools()
            # 打印所有工具名称
            print("[Tools]", [t.name for t in tools.tools])
+           # 调用名为 "hello" 的工具，传入空参数
+           result = await session.call_tool("hello", arguments={})
+           # 用于收集文本返回内容
+           tool_texts = []
+           # 遍历每个内容块
+           for block in result.content:
+               # 如果内容块是字典且类型为"text"，提取 "text" 字段
+               if isinstance(block, dict) and block.get("type") == "text":
+                   tool_texts.append(block.get("text", ""))
+               # 如果内容块有 text 属性，直接取出 text
+               elif hasattr(block, "text"):
+                   tool_texts.append(block.text)
+           # 打印调用 hello 工具的响应内容
+           print("[Call hello]", " | ".join(tool_texts) or str(result.structured_content))

# 定义以 stdio 运行的服务器端函数
def run_server_stdio():
    # 在标准错误流打印启动提示
    print("启动 MCP 服务器（stdio 模式）...", file=sys.stderr)
    # 以 stdio 方式运行 mcp 服务器
    mcp.run(transport="stdio")

# 主入口函数，根据命令行参数决定运行模式
def main():
    # 如果参数含有"serve"，以服务器模式运行
    if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        run_server_stdio()
    else:
        # 否则启动客户端测试流程，自动拉起服务端
        print("启动 MCP 客户端测试...")
        print("将自动启动服务器进程并测试连接")
        asyncio.run(run_client())

# 判断是否直接运行此文件
if __name__ == "__main__":
    main()

2.2. session.py #

mcp_lite/client/session.py

import sys
# 导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.types 模块导入相关类型和常量
from mcp_lite.types import (
      InitializeRequestParams,    # 导入初始化请求参数结构体
      InitializeRequest,          # 导入初始化请求结构体
      LATEST_PROTOCOL_VERSION,    # 导入协议最新版本号
      ClientCapabilities,         # 导入客户端能力描述
      Implementation,             # 导入实现信息结构体
      InitializedNotification,    # 导入初始化完成通知
      InitializeResult,           # 导入初始化请求返回结构体
      JSONRPCRequest,             # 导入 JSONRPC 请求类型
      JSONRPCResponse,            # 导入 JSONRPC 响应类型
      JSONRPCError,               # 导入 JSONRPC 错误类型
      JSONRPCNotification,        # 导入 JSONRPC 通知类型
      ListToolsRequest,           # 导入获取工具列表请求
      ListToolsResult,            # 导入获取工具列表响应结构体
+     CallToolResult,             # 导入调用工具返回结果的数据结构
+     CallToolRequest,            # 导入调用工具请求的数据结构
+     CallToolRequestParams,      # 导入调用工具请求参数的数据结构
)
# 定义 MCPError 异常类，用于表示协议相关错误
class MCPError(Exception):
    # 初始化方法，接收错误码、错误信息和可选的数据
    def __init__(self, code, message, data=None):
        # 将错误码、错误信息、附加数据赋值为实例属性
        self.code, self.message, self.data = code, message, data
        # 调用父类 Exception 的初始化，只传递错误信息
        super().__init__(message)

    # 类方法，利用 JSONRPCError 对象创建 MCPError 实例
    @classmethod
    def from_jsonrpc(cls, err):
        # 从 JSONRPCError.error 对象中取出 code/message/data 创建 MCPError
        return cls(err.error.code, err.error.message, err.error.data)
# 定义 ClientSession 类，用于描述客户端会话
class ClientSession:
    # 构造方法，接收读取流和写入流
    def __init__(self, read_stream, write_stream):
        # 赋值读取流和写入流
        self._read, self._write = read_stream, write_stream
        # 初始化请求 id 计数器为 0
        self._req_id = 0
    # 内部方法，发送请求并同步等待响应
    def _request(self, req):
        # 当前请求 id
        rid = self._req_id
        # 自增请求 id，确保下次唯一
        self._req_id += 1
        # 对请求对象进行序列化，转为 dict
        d = req.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 构建 JSONRPCRequest，再封装入 SessionMessage，通过写入流发送
        jreq = JSONRPCRequest(jsonrpc="2.0", id=rid, method=d["method"], params=d.get("params"))
        # 打印请求信息到标准错误流
        print("[Client] Request:", jreq.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
        # 将请求对象封装成 SessionMessage 后发送
        self._write.send(SessionMessage(message=jreq))
        # 循环等待响应
        while True:
            # 从读取流取出一条消息
            msg = self._read.get()
            # 若取到 None，说明连接断开，抛出异常
            if msg is None:
                raise MCPError(-32000, "Connection closed")
            # 若消息类型不是 SessionMessage，忽略继续等
            if not isinstance(msg, SessionMessage):
                continue
            # 取消息内容
            m = msg.message
            # 若为 Response 或 Error，且 id 匹配
            if isinstance(m, (JSONRPCResponse, JSONRPCError)) and getattr(m, "id", None) == rid:
                print("[Client] Response:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
                # 若为错误，抛出 MCPError 异常
                if isinstance(m, JSONRPCError):
                    raise MCPError.from_jsonrpc(m)
                # 为正常响应则返回结果数据
                return m.result
    # 内部方法，发送通知消息
    def _notify(self, n):
        # 通知对象序列化为 dict
        d = n.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 封装为 JSONRPCNotification，再封装成 SessionMessage 后发出
        self._write.send(SessionMessage(
            message=JSONRPCNotification(
                jsonrpc="2.0",
                method=d["method"],
                params=d.get("params"),
            )
        ))            
    # 初始化过程，发送初始化请求，收到响应后再发初始化完成通知
    def initialize(self):
        # 构造初始化请求，并发送等待返回
        r = self._request(InitializeRequest(
            params=InitializeRequestParams(
                protocol_version=LATEST_PROTOCOL_VERSION,         # 协议版本号
                capabilities=ClientCapabilities(),                # 客户端能力
                client_info=Implementation(name="mcp_lite", version="0.1.0"),  # 客户端信息
            )
        ))
        # 发送"初始化完成"通知
        self._notify(InitializedNotification())
        # 用 InitializeResult 验证响应并返回
        return InitializeResult.model_validate(r, by_name=False)
        # 获取工具列表
    def list_tools(self):
        # 发送 ListToolsRequest 请求，使用 ListToolsResult 校验并返回
        return ListToolsResult.model_validate(self._request(ListToolsRequest()), by_name=False)    
     # 调用指定工具方法，传入工具名和参数
+   def call_tool(self, name, arguments=None):
        # 构造 CallToolRequest 并发送，请求参数包含工具名和参数
+       return CallToolResult.model_validate(
+           self._request(CallToolRequest(
+               params=CallToolRequestParams(name=name, arguments=arguments or {})
+           )),
+           by_name=False,
+       )

2.3. fastmcp.py #

mcp_lite/server/fastmcp.py

# 导入 sys 模块，用于标准输入输出操作
import sys
# 导入 inspect 用于函数签名分析
import inspect
# 导入 asyncio 用于异步操作
import asyncio
# 从 mcp_lite.message 模块导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.server 模块导入 stdio 子模块
from mcp_lite.server import stdio
# 从 mcp_lite.types 模块导入所有相关类型和类
from mcp_lite.types import (
     JSONRPCRequest,            # JSONRPC 请求类型
     JSONRPCError,              # JSONRPC 错误类型
     ErrorData,                 # 错误数据类型
     InitializeResult,          # 初始化响应类型
     LATEST_PROTOCOL_VERSION,   # 最新协议版本常量
     ToolsCapability,           # 工具能力类型
     ServerCapabilities,        # 服务器功能特性类型
     Implementation,            # 实现信息类型
     JSONRPCResponse,           # JSONRPC 响应类型
     ListToolsResult,           # 工具列表响应类型
     Tool,                      # 工具类型
+    CallToolRequestParams,     # 工具调用请求参数类型
+    CallToolResult,            # 工具调用结果类型
)
# 内部函数：根据函数定义提取参数信息并生成 schema
def _schema(fn):
    # 获取函数签名
    sig = inspect.signature(fn)
    # 用于存储参数属性的字典
    props = {}
    # 用于存储必需参数名的列表
    req = []
    # 遍历所有参数
    for n, p in sig.parameters.items():
        # 跳过以 "_" 开头的参数
        if n.startswith("_"):
            continue
        # 所有参数类型都设为 string，title 为参数名
        props[n] = {"type": "string", "title": n}
        # 如果参数没有默认值，则为必需参数
        if p.default is inspect.Parameter.empty:
            req.append(n)
    # 返回对象类型 schema，包括属性和必需项目
    return {"type": "object", "properties": props, "required": req}
# 内部类：用于封装工具函数
class _Tool:
    # 初始化方法，接收目标函数、工具名和描述
    def __init__(self, fn, name=None, desc=None):
        # 保存原始函数引用
        self.fn = fn
        # 工具名称，优先采用传参，否则用函数名
        self.name = name or fn.__name__
        # 工具描述，优先采用传参，否则用函数 docstring
        self.desc = desc or (fn.__doc__ or "").strip()
        # 自动生成入参 schema
        self.schema = _schema(fn)
        # 判断函数是否为异步（协程）函数
        self.async_fn = asyncio.iscoroutinefunction(fn)

    # 转换为 Tool 类型的对象
    def to_tool(self):
        # 返回 Tool 类型实例
        return Tool(name=self.name, description=self.desc or None, input_schema=self.schema)

    # 执行工具函数，参数为字典类型
    def run(self, args):
        # 如果是异步函数则使用 asyncio.run 执行，否则同步调用
        if self.async_fn:
            return asyncio.run(self.fn(**args))
        return self.fn(**args)
# 主服务类 FastMCP
class FastMCP:
    # 初始化方法，支持自定义服务器名称
    def __init__(self, name="mcp-server"):
        # 保存服务器名称
        self.name = name
        # 初始化工具池为字典
        self._tools = {}
    # 工具注册装饰器，支持自定义工具名和描述
    def tool(self, name=None, description=None):
        # 装饰器实际函数
        def deco(fn):
            # 创建 _Tool 实例
            t = _Tool(fn, name, description)
            # 注册进工具池
            self._tools[t.name] = t
            # 返回原始函数
            return fn
        # 返回装饰器
        return deco    
     # 实际 JSONRPC 方法分发与业务处理
    def _handle(self, req):
        # 拆分 method、params、id
        method, params, rid = req.method, req.params or {}, req.id

        # 处理初始化请求
        if method == "initialize":
            # 组装初始化响应，包括协议版本、能力与服务信息
            r = InitializeResult(
                protocol_version=LATEST_PROTOCOL_VERSION,
                capabilities=ServerCapabilities(tools=ToolsCapability()),
                server_info=Implementation(name=self.name, version="0.1.0"),
            )
            # 返回 JSONRPCResponse 包装的结果
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
        # 处理工具列表请求
        if method == "tools/list":
            # 构建工具列表结果
            r = ListToolsResult(tools=[t.to_tool() for t in self._tools.values()])
            # 返回 JSONRPCResponse 包装的结果
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))  
                # 处理工具调用请求
+       if method == "tools/call":
            # 参数校验及实例化为 CallToolRequestParams
+           p = CallToolRequestParams.model_validate(params, by_name=False)
            # 查找对应工具
+           t = self._tools.get(p.name)
            # 找不到工具时返回参数错误
+           if not t:
+               return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32602, message=f"Unknown tool: {p.name}"))
+           try:
                # 执行工具函数
+               out = t.run(p.arguments or {})
                # 如果结果已经是 CallToolResult，直接拆取 content
+               if isinstance(out, CallToolResult):
+                   c = out.content
                # 返回字符串时，封装为文本类型内容列表
+               elif isinstance(out, str):
+                   c = [{"type": "text", "text": out}]
                # 结果为 None 时，返回空内容列表
+               elif out is None:
+                   c = []
                # 其它类型结果统一转为字符串文本
+               else:
+                   c = [{"type": "text", "text": str(out)}]
                # 构造 CallToolResult 作为最终响应体
+               r = CallToolResult(content=c)
+           except Exception as e:
                # 工具执行出错时，返回错误内容、标记 is_error
+               r = CallToolResult(content=[{"type": "text", "text": str(e)}], is_error=True)
            # 返回 JSONRPCResponse 包装的工具执行结果
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))      
        # 不支持的方法，返回 "方法未找到" 错误（code -32601）
        return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32601, message=f"Method not found: {method}"))    
    # 处理 SessionMessage 消息，主要路由 JSONRPCRequest
    def _handle_msg(self, msg):
        # 判断消息类型不是 SessionMessage 则忽略
        if not isinstance(msg, SessionMessage):
            return None
        # 获取消息体
        m = msg.message
        # 只处理 JSONRPCRequest 类型
        if not isinstance(m, JSONRPCRequest):
            return None
        print("[Server] Request:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)    
        try:
            # 调用实际业务处理方法
            resp = self._handle(m)
            print("[Server] Response:", resp.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return resp
        except Exception as e:
            # 捕获异常，返回标准 JSONRPCError（code -32603，内部错误）
            err = JSONRPCError(jsonrpc="2.0", id=m.id, error=ErrorData(code=-32603, message=str(e)))
            print("[Server] Response (error):", err.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return err
    # 运行服务方法，默认采用 stdio 传输方式
    def run(self, transport="stdio"):
        # 仅支持 stdio，其他方式抛出异常
        if transport != "stdio":
            raise ValueError(f"unsupported transport: {transport}")
        # 启动 stdio server，当前对象 _handle_msg 作为消息处理回调
        stdio.stdio_server(self._handle_msg)    

2.4. types.py #

mcp_lite/types.py

# 导入 Any 和 Literal 类型注解
from typing import Any, Literal
# 从 pydantic 导入 BaseModel、ConfigDict、Field、TypeAdapter
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
# 导入 pydantic 的 to_camel 驼峰命名生成器
from pydantic.alias_generators import to_camel
# 定义 RequestId 类型，既可以是 int 也可以是 str
RequestId = int | str
# 定义 MCP 的基础模型类，支持驼峰命名和按名称填充
class MCPModel(BaseModel):
    # 指定 Pydantic 模型配置：使用驼峰形式命名和按名称填充
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
# 定义客户端能力结构体
class ClientCapabilities(MCPModel):
    # 可选字段：客户端实验性能力扩展，键为 str，值为字典
    experimental: dict[str, dict[str, Any]] | None = None
# 定义服务器或客户端的实现信息结构体
class Implementation(MCPModel):
    # 实现的名称
    name: str = ""
    # 实现的版本号
    version: str = ""
    # 可选字段：人类可读的标题
    title: str | None = None
    # 可选字段：实现的描述
    description: str | None = None    
# 定义初始化请求参数结构体
class InitializeRequestParams(MCPModel):
    # 协议版本号
    protocol_version: str = ""
    # 可选字段：客户端能力描述
    capabilities: ClientCapabilities = None
    # 可选字段：客户端实现信息
    client_info: Implementation = None
# 定义初始化请求结构体
class InitializeRequest(MCPModel):
    # 方法名，固定为 "initialize"
    method: Literal["initialize"] = "initialize"
    # 可选字段：参数，类型为 InitializeRequestParams
    params: InitializeRequestParams = None    
# 当前协议的最新版本号
LATEST_PROTOCOL_VERSION = "2024-11-05"
# 定义工具相关能力结构体
class ToolsCapability(MCPModel):
    # 工具列表是否发生变化，可选布尔类型
    list_changed: bool | None = None
# 定义服务端能力描述结构体
class ServerCapabilities(MCPModel):
    # 可选字段：实验性能力扩展
    experimental: dict[str, dict[str, Any]] | None = None
    # 可选字段：工具能力
    tools: ToolsCapability | None = None
# 定义初始化响应结构体
class InitializeResult(MCPModel):
    # 协议版本号
    protocol_version: str = ""
    # 可选字段：服务端能力描述
    capabilities: ServerCapabilities = None
    # 可选字段：服务端实现信息
    server_info: Implementation = None
    # 可选字段：初始化说明信息
    instructions: str | None = None
# 定义客户端初始化完成通知结构体
class InitializedNotification(MCPModel):
    # 方法名，固定为 "notifications/initialized"
    method: Literal["notifications/initialized"] = "notifications/initialized"
    # 可选字段：通知参数，可以为字典或 None
    params: dict[str, Any] | None = None    
# 定义 JSONRPC 请求的数据结构
class JSONRPCRequest(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 请求 ID，可以为 int 或 str 类型
    id: RequestId = None
    # 方法名称，字符串类型
    method: str = ""
    # 方法参数，为一个字典或 None
    params: dict[str, Any] | None = None    

# 定义 JSONRPC 响应的数据结构
class JSONRPCResponse(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 响应的 ID，需要与请求 ID 匹配
    id: RequestId = None
    # 响应结果，可以为字典或 None
    result: dict[str, Any] = None   
# 定义错误的数据结构
class ErrorData(BaseModel):
    # 错误码，默认值为 0
    code: int = 0
    # 错误信息，默认值为空字符串
    message: str = ""
    # 附加的错误数据，可以为任意类型或 None
    data: Any = None     
# 定义 JSONRPC 错误消息的数据结构
class JSONRPCError(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 错误对应的请求 ID，可以为 None
    id: RequestId | None = None
    # 错误的详细信息，类型为 ErrorData
    error: ErrorData = None    
# 定义 JSONRPC 通知的数据结构（没有 id 字段）
class JSONRPCNotification(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 通知的方法名称
    method: str = ""
    # 通知参数，可以为字典或者 None
    params: dict[str, Any] | None = None    

# 定义所有 JSONRPC 消息的联合类型
JSONRPCMessage = JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError

# 定义 JSONRPC 消息适配器，用于类型自动推断和校验
jsonrpc_message_adapter = TypeAdapter(JSONRPCMessage)
# 定义工具描述数据结构体
class Tool(MCPModel):
    # 工具名称
    name: str = ""
    # 可选字段：工具描述
    description: str | None = None
    # 工具输入参数的 schema，默认为空字典
    input_schema: dict[str, Any] = Field(default_factory=dict)
    # 可选字段：工具输出 schema
    output_schema: dict[str, Any] | None = None
# 定义获取工具列表请求结构体
class ListToolsRequest(MCPModel):
    # 方法名，固定为 "tools/list"
    method: Literal["tools/list"] = "tools/list"
    # 可选字段：参数，可以为字典或 None
    params: dict[str, Any] | None = None

# 定义获取工具列表响应结构体
class ListToolsResult(MCPModel):
    # 工具列表，默认为空列表
    tools: list[Tool] = []
    # 可选字段：分页游标，可为 None
    next_cursor: str | None = None

# 定义获取工具列表响应结构体
class ListToolsResult(MCPModel):
    # 工具列表，默认为空列表
    tools: list[Tool] = []
    # 可选字段：分页游标，可为 None
    next_cursor: str | None = None
# 定义调用工具请求参数结构体
+class CallToolRequestParams(MCPModel):
    # 工具名称
+   name: str = ""
    # 可选字段：输入参数，为字典类型或 None
+   arguments: dict[str, Any] | None = None        
# 定义调用工具请求结构体
+class CallToolRequest(MCPModel):
    # 方法名，固定为 "tools/call"
+   method: Literal["tools/call"] = "tools/call"
    # 可选字段：参数，类型为 CallToolRequestParams
+   params: CallToolRequestParams = None
# 定义调用工具响应结构体
+class CallToolResult(MCPModel):
    # 返回内容，为一个列表，默认为空
+   content: list = Field(default_factory=list)
    # 可选字段：结构化内容，为字典或 None
+   structured_content: dict[str, Any] | None = None
    # 是否为错误，布尔类型，默认 False
+   is_error: bool = False    

2.5 工作流程 #

2.5.1 整体功能 #

在 tools/list 之后，增加 调用工具 的能力：

1. 客户端：通过 session.call_tool(name, arguments) 调用指定工具
2. 服务端：处理 tools/call 请求，执行对应工具函数并返回结果
3. 协议：新增 tools/call 方法，请求包含工具名和参数，响应为 CallToolResult

2.5.2 1.mcp.py 客户端调用 hello 工具 #

# 调用名为 "hello" 的工具，参数为空字典
result = session.call_tool("hello", arguments={})
# 用于存放从 content 中提取出的文本内容
tool_texts = []
# 遍历工具返回的内容块列表
for block in result.content:
    # 如果内容块是字典且 "type" 字段为 "text"
    if isinstance(block, dict) and block.get("type") == "text":
        # 提取字典中的 "text" 字段内容，添加到 tool_texts
        tool_texts.append(block.get("text", ""))
    # 如果内容块是一个拥有 text 属性的对象
    elif hasattr(block, "text"):
        # 取出对象的 text 属性，添加到 tool_texts
        tool_texts.append(block.text)
# 输出调用 hello 工具的结果，多个文本用 " | " 拼接；若无文本则用 structured_content 的字符串表示
print("[Call hello]", " | ".join(tool_texts) or str(result.structured_content))

• session.call_tool("hello", arguments={})：调用名为 hello 的工具，参数为空字典
• result 是 CallToolResult，包含 content 和 structured_content
• content 是内容块列表，每个块可以是：
  • 字典：{"type": "text", "text": "..."}
  • 对象：有 text 属性
• 遍历 content 提取文本，用 " | " 拼接；若没有文本则用 structured_content 的字符串表示

2.5.3 2.2 session.py 新增 call_tool #

+     CallToolResult,             # 导入调用工具返回结果的数据结构
+     CallToolRequest,            # 导入调用工具请求的数据结构
+     CallToolRequestParams,      # 导入调用工具请求参数的数据结构

# 定义 call_tool 方法，用于调用指定的工具
def call_tool(self, name, arguments=None):
    # 调用 _request 方法，发送 CallToolRequest，请求参数为工具名和参数字典
    return CallToolResult.model_validate(
        self._request(
            CallToolRequest(
                # 构造 CallToolRequestParams，工具名为 name，参数为 arguments（若为 None 则用空字典）
                params=CallToolRequestParams(name=name, arguments=arguments or {})
            )
        ),
        # 校验模型时不按名称映射（by_name=False）
        by_name=False,
    )

• call_tool(name, arguments)：构造 CallToolRequest，params 为 CallToolRequestParams(name=name, arguments=arguments or {})
• 通过 _request 发送并等待 JSON-RPC 响应
• 用 CallToolResult.model_validate 校验并返回

2.5.4 2.3 fastmcp.py处理 tools/call #

2.5.5 新增导入 #

+    CallToolRequestParams,     # 工具调用请求参数类型
+    CallToolResult,            # 工具调用结果类型

2.5.6 处理 tools/call 请求 #

# 判断请求方法是否为 "tools/call"
if method == "tools/call":
    # 解析参数，将 params 转换为 CallToolRequestParams 对象
    p = CallToolRequestParams.model_validate(params, by_name=False)
    # 根据工具名称查找已注册的工具
    t = self._tools.get(p.name)
    # 工具未找到时，返回 JSONRPCError 错误，提示未知工具
    if not t:
        return JSONRPCError(..., message=f"Unknown tool: {p.name}")
    try:
        # 执行工具函数，将参数字典传递给工具
        out = t.run(p.arguments or {})
        # 如果返回值是 CallToolResult，直接使用其 content
        if isinstance(out, CallToolResult):
            c = out.content
        # 如果返回值是字符串，将其封装为标准文本内容块
        elif isinstance(out, str):
            c = [{"type": "text", "text": out}]
        # 如果返回值为 None，内容块列表为空
        elif out is None:
            c = []
        # 其他类型的返回值，统一转为字符串作为文本内容块
        else:
            c = [{"type": "text", "text": str(out)}]
        # 构造 CallToolResult 结果对象，content 为上述内容列表
        r = CallToolResult(content=c)
    except Exception as e:
        # 工具执行异常时，封装错误信息为文本内容块，并标记 is_error=True
        r = CallToolResult(content=[{"type": "text", "text": str(e)}], is_error=True)
    # 返回 JSONRPCResponse，result 为 CallToolResult 的字典形式
    return JSONRPCResponse(..., result=r.model_dump(...))

流程：

1. 用 CallToolRequestParams.model_validate(params) 解析请求参数
2. 用 self._tools.get(p.name) 查找工具，找不到则返回 Unknown tool 错误
3. 调用 t.run(p.arguments or {}) 执行工具
4. 根据 out 类型构造 content：
   • CallToolResult：直接用其 content
   • str：[{"type": "text", "text": out}]
   • None：[]
   • 其他：[{"type": "text", "text": str(out)}]
5. 异常时：CallToolResult(content=[...], is_error=True)
6. 返回 JSON-RPC 响应，result 为 CallToolResult 的字典形式

2.5.7 2.4 types.py工具调用相关类型 #

2.5.8 CallToolRequestParams #

# 定义工具调用请求参数的数据结构类
class CallToolRequestParams(MCPModel):
    # 工具名称
    name: str = ""
    # 调用参数，类型为字典（可为 None）
    arguments: dict[str, Any] | None = None

2.5.9 CallToolRequest #

# 定义工具调用请求的数据结构类
class CallToolRequest(MCPModel):
    # 方法名，固定为 "tools/call"
    method: Literal["tools/call"] = "tools/call"
    # 请求参数，类型为 CallToolRequestParams，可为 None
    params: CallToolRequestParams = None

2.5.10 CallToolResult #

# 定义工具调用结果的数据结构
class CallToolResult(MCPModel):
    # 内容块列表，默认为空列表
    content: list = Field(default_factory=list)
    # 可选的结构化内容，类型为字典或None
    structured_content: dict[str, Any] | None = None
    # 是否为错误结果，默认为False
    is_error: bool = False

• content：MCP 标准的内容块列表，常见格式为 {"type": "text", "text": "..."}
• structured_content：可选的结构化数据
• is_error：为 True 表示工具执行出错

2.5.11 消息流示意 #

客户端                              服务端
  │                                    │
  │  CallToolRequest                   │
  │  method: "tools/call"              │
  │  params: {name: "hello", arguments: {}}  │
  │ ─────────────────────────────────►│
  │                                    │  _handle("tools/call")
  │                                    │  查找 _tools["hello"]
  │                                    │  执行 hello()
  │                                    │  返回 "Hello from MCP server!"
  │  CallToolResult                    │
  │  content: [{"type":"text","text":"Hello from MCP server!"}]
  │ ◄─────────────────────────────────│
  │                                    │
  │  result.content → 提取文本 → 打印   │

2.5.12 返回值到 content 的映射 #

2.5.13 小结 #

整体上，这部分实现了从「列出工具」到「调用工具并获取结果」的完整流程。