5.1_API参考

5.1 API参考

本章提供MCP协议的完整API参考文档，包括服务器API和客户端API的详细说明。这些API参考将帮助开发者更深入地理解MCP的各个组件，并在构建自己的应用程序时作为查阅依据。

目录

服务器API

FastMCP类

FastMCP是MCP服务器的主要入口点，用于创建和配置MCP服务器。

from mcp.server.fastmcp import FastMCP

# 创建一个MCP服务器实例
mcp = FastMCP(
    name="My MCP Server",            # 服务器名称
    description="My MCP服务器描述",   # 可选的服务器描述
    version="1.0.0",                 # 可选的版本号
    verbose=True                     # 可选的详细日志输出设置
)

核心方法

方法 描述
run(transport="stdio") 启动MCP服务器，使用指定的传输方式
resource(uri_pattern) 用于注册资源的装饰器
tool() 用于注册工具的装饰器
prompt() 用于注册提示的装饰器

支持的传输方式

传输方式 描述
"stdio" 通过标准输入/输出进行通信（默认）
"sse" 使用Server-Sent Events进行通信
"websocket" 使用WebSocket进行通信
"http" 使用HTTP API进行通信

资源装饰器

资源装饰器用于将函数注册为MCP资源。

@mcp.resource(uri_pattern)
def get_resource(param1, param2, ...) -> Any:
    """资源描述"""
    # 实现...

参数

参数 类型 描述
uri_pattern str 资源的URI模式，支持参数化路径

示例 

@mcp.resource("data://users/{user_id}")
def get_user_data(user_id: str) -> dict:
    """获取用户数据"""
    # 实现...

工具装饰器

工具装饰器用于将函数注册为MCP工具。

@mcp.tool()
def my_tool(param1: Type1, param2: Type2, ...) -> ReturnType:
    """工具描述"""
    # 实现...

参数

工具装饰器的参数：

参数 类型 描述
name str 可选的工具名称，默认使用函数名
description str 可选的工具描述，默认使用函数文档字符串

示例 

@mcp.tool(name="calculator")
def calculate(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

提示装饰器

提示装饰器用于将函数注册为MCP提示模板。

@mcp.prompt()
def my_prompt(param1: Type1, param2: Type2, ...) -> Union[str, List[Message]]:
    """提示描述"""
    # 实现...

参数

参数 类型 描述
name str 可选的提示名称，默认使用函数名
description str 可选的提示描述，默认使用函数文档字符串

示例 

@mcp.prompt()
def code_review(code: str) -> str:
    """代码审查提示"""
    return f"""
    请审查以下代码，指出潜在的问题和改进建议：
    
    ```
    {code}
    ```
    """

上下文对象

上下文对象在服务器端提供对当前会话和请求的访问。

from mcp.server.fastmcp import Context

@mcp.tool()
def context_aware_tool(ctx: Context, param1: str) -> str:
    """使用上下文的工具"""
    # 访问会话ID
    session_id = ctx.session_id
    
    # 记录日志
    ctx.logger.info(f"工具被调用，参数: {param1}")
    
    # 实现...

属性和方法

属性/方法 描述
session_id 当前会话的唯一标识符
logger 日志对象，用于记录日志
state 会话状态字典，用于存储会话特定数据
get_tool(name) 获取指定名称的工具
get_resource(uri) 获取指定URI的资源
get_prompt(name) 获取指定名称的提示

服务器生命周期方法

服务器生命周期方法用于控制服务器的初始化和关闭行为。

@mcp.on_initialize
async def initialize(ctx: Context) -> None:
    """服务器初始化时调用"""
    ctx.logger.info("服务器正在初始化...")
    # 执行初始化操作

@mcp.on_shutdown
async def shutdown(ctx: Context) -> None:
    """服务器关闭时调用"""
    ctx.logger.info("服务器正在关闭...")
    # 执行清理操作

客户端API

ClientSession类

ClientSession是MCP客户端的主要接口，用于与MCP服务器交互。

from mcp import ClientSession

async def main():
    async with ClientSession(read, write) as session:
        # 使用会话...

核心方法

方法 描述
initialize() 初始化与服务器的连接
close() 关闭与服务器的连接
list_resources() 列出可用的资源
list_tools() 列出可用的工具
list_prompts() 列出可用的提示
read_resource(uri, arguments) 读取资源
call_tool(name, arguments) 调用工具
render_prompt(name, arguments) 渲染提示

连接方法

MCP提供多种连接方式，适用于不同的场景。

Stdio连接 

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

async def run():
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 使用会话...

SSE连接 

from mcp import ClientSession
from mcp.client.sse import sse_client

async def run():
    async with sse_client("http://localhost:8000/mcp") as (read, write):
        async with ClientSession(read, write) as session:
            # 使用会话...

WebSocket连接 

from mcp import ClientSession
from mcp.client.websocket import websocket_client

async def run():
    async with websocket_client("ws://localhost:8000/mcp") as (read, write):
        async with ClientSession(read, write) as session:
            # 使用会话...

资源访问 

# 列出所有资源
resources = await session.list_resources()

# 读取资源
user_data = await session.read_resource(
    "data://users/123",
    arguments={"include_details": True}
)

工具调用 

# 列出所有工具
tools = await session.list_tools()

# 调用工具
result = await session.call_tool(
    "calculator",
    arguments={"expression": "2 + 2"}
)

提示使用 

# 列出所有提示
prompts = await session.list_prompts()

# 渲染提示
prompt = await session.render_prompt(
    "code_review",
    arguments={"code": "def hello(): print('world')"}
)

数据类型和接口

基础类型

Response

所有MCP请求的通用响应结构。

class Response:
    meta: Optional[Dict[str, Any]]  # 元数据，如分页信息
    
class ResourceResponse(Response):
    content: Union[List[Content], Content]  # 资源内容
    is_error: bool  # 是否是错误响应
    
class ToolResponse(Response):
    content: Union[List[Content], Content]  # 工具执行结果
    is_error: bool  # 是否是错误响应
    
class PromptResponse(Response):
    content: Union[str, List[Message]]  # 渲染后的提示
    is_error: bool  # 是否是错误响应

Content

内容类型，用于表示资源或工具的返回值。

class TextContent:
    type: Literal["text"]  # 内容类型
    text: str  # 文本内容
    annotations: Optional[List[Annotation]]  # 可选的注释
    
class JsonContent:
    type: Literal["json"]  # 内容类型
    json: Any  # JSON数据
    
class ImageContent:
    type: Literal["image"]  # 内容类型
    image_data: bytes  # 图像数据
    mime_type: str  # MIME类型

资源类型 

class Resource:
    uri: str  # 资源URI
    description: Optional[str]  # 资源描述
    arguments: Optional[Dict[str, Any]]  # 参数模式

工具类型 

class Tool:
    name: str  # 工具名称
    description: Optional[str]  # 工具描述
    arguments: Optional[Dict[str, Any]]  # 参数模式

提示类型 

class Prompt:
    name: str  # 提示名称
    description: Optional[str]  # 提示描述
    arguments: Optional[Dict[str, Any]]  # 参数模式
    
class Message:
    role: str  # 消息角色 (user, assistant, system)
    content: str  # 消息内容

错误类型 

class ErrorContent:
    type: Literal["error"]  # 内容类型
    status_code: int  # 状态码
    reason: str  # 错误原因
    detail: Optional[str]  # 详细错误信息
    help_text: Optional[str]  # 帮助文本

小结

本章详细介绍了MCP的服务器API和客户端API，以及相关的数据类型和接口。通过这些API，开发者可以创建自定义的MCP服务器，实现资源访问、工具调用和提示渲染，从而为大语言模型提供丰富的上下文和功能。

在实际开发中，请参考最新的官方文档和API参考，因为MCP协议可能会随着时间的推移而更新和改进。

下一章将介绍MCP的错误代码与解决方案，帮助您诊断和解决在使用MCP时可能遇到的常见问题。

© 2025 Lyrical Wander
使用 Hugo 构建
主题 Stack 由 Jimmy 设计