MCP 应用程序接口参考

本文档提供 AG-Kit Python 模型上下文协议(MCP)集成组件的完整应用程序接口参考。

核心类

AGKitMCPServer

将 AG-Kit 工具作为标准 MCP 服务器暴露。

构造函数

AGKitMCPServer(config: dict)

参数:

config: dict - 服务器配置字典

配置字典:

{
    'name': str,                    # 服务器名称（必填）
    'version': str,                 # 服务器版本（必填）
    'description': str,             # 可选的服务器描述
    'log_level': str,               # 日志级别：'info', 'debug'（默认：'info'）
    'capabilities': dict,           # 可选的 MCP 能力
    'error_handling': str,          # 'throw' 或 'return_error'（默认：'return_error'）
}

方法

`register_tool()`

向服务器注册单个 AG-Kit 工具。

register_tool(tool: BaseTool, registration: dict = None) -> dict

参数:

tool: BaseTool - 要注册的 AG-Kit 工具
registration: dict - 可选的工具注册配置

注册配置:

{
    'name': str,           # 覆盖工具名称
    'description': str,    # 覆盖工具描述
    'name_prefix': str,    # 为工具名称添加前缀
}

返回: dict - 已注册工具的元数据

`register_tools()`

使用共享配置注册多个 AG-Kit 工具。

register_tools(tools: list[BaseTool], registration: dict = None) -> list[dict]

参数:

tools: list[BaseTool] - 要注册的 AG-Kit 工具列表
registration: dict - 可选的共享配置

返回: list[dict] - 已注册工具的元数据列表

`unregister_tool()`

从服务器移除工具。

unregister_tool(name: str) -> bool

参数:

name: str - 要移除的工具名称

返回: bool - 如果工具被移除返回 True，未找到返回 False

`run()`

使用指定传输方式启动 MCP 服务器。

async def run(transport_config: dict) -> None

参数:

transport_config: dict - 传输配置

传输配置类型:

# 标准输入输出传输
{
    'type': 'stdio'
}

# 内存传输
{
    'type': 'memory',
    'memory_id': str  # 可选，默认为 'default'
}

# SSE 传输
{
    'type': 'sse',
    'sse_setup': Callable  # 异步设置函数
}

# 可流式 HTTP 传输
{
    'type': 'streamableHttp',
    'streamable_http_setup': Callable  # 异步设置函数
}

`stop()`

停止 MCP 服务器并清理资源。

async def stop() -> None

`call_tool()`

直接调用已注册工具。

async def call_tool(name: str, args: dict) -> dict

参数:

name: str - 要执行的工具名称
args: dict - 工具参数

返回: dict - MCP 格式的工具结果

`list_tools()`

获取所有已注册工具的列表。

def list_tools() -> list[dict]

返回: 工具元数据字典列表

`is_server_running()`

检查服务器当前是否正在运行。

def is_server_running() -> bool

返回: bool - 如果服务器正在运行返回 True

`get_stats()`

获取服务器统计信息和状态。

def get_stats() -> dict

返回:

{
    'total_tools': int,
    'tool_names': list[str],
    'tool_types': dict[str, int],
    'is_running': bool,
    'server_info': {
        'name': str,
        'version': str,
        'description': str  # 可选
    }
}

`get_server_info()`

获取服务器信息。

def get_server_info() -> dict

返回:

{
    'name': str,
    'version': str,
    'description': str,  # 可选
    'tool_count': int,
    'tools': list[str]
}

`get_config()`

获取当前服务器配置。

def get_config() -> dict

`update_config()`

更新服务器配置。

def update_config(new_config: dict) -> None

`get_connection_info()`

获取连接信息。

def get_connection_info() -> dict

返回:

{
    'is_running': bool,
    'transport_type': str | None,
    'memory_id': str | None
}

`get_transport_type()`

获取当前传输类型。

def get_transport_type() -> str | None

`get_memory_id()`

如果使用内存传输，获取内存传输 ID。

def get_memory_id() -> str | None

`add_event_listener()`

添加服务器事件监听器。

def add_event_listener(listener: Callable) -> None

参数:

listener: Callable - 接收 MCPEvent 的事件监听函数

`remove_event_listener()`

移除事件监听器。

def remove_event_listener(listener: Callable) -> bool

返回: bool - 如果监听器被移除返回 True

MCPClientManager

管理外部 MCP 服务器的连接。

构造函数

MCPClientManager()

方法

`add_server()`

添加并连接到 MCP 服务器。

async def add_server(
    server_id: str,
    config: dict,
    connection_options: dict = None
) -> None

参数:

server_id: str - 服务器的唯一标识符
config: dict - 客户端配置
connection_options: dict - 可选的连接设置

客户端配置:

{
    'name': str,              # 客户端名称（必填）
    'version': str,           # 客户端版本（必填）
    'transport': dict,        # 传输配置（如果未提供客户端则为必填）
    'client': Client,         # 预连接客户端（可选）
    'capabilities': dict,     # 可选的能力
}

连接选项:

{
    'auto_reconnect': bool,         # 启用自动重连（默认：True）
    'reconnect_delay': int,         # 重连尝试之间的延迟（默认：5000ms）
    'max_reconnect_attempts': int,  # 最大重连尝试次数（默认：3）
    'heartbeat_interval': int,      # 心跳间隔（默认：30000ms）
}

`disconnect_server()`

断开与特定服务器的连接。

async def disconnect_server(server_id: str) -> None

`disconnect_all()`

断开与所有服务器的连接。

async def disconnect_all() -> None

`create_client_tools()`

为所有可用的 MCP 工具创建 AG-Kit 工具包装器。

def create_client_tools(server_id: str = None) -> list[MCPClientTool]

参数:

server_id: str - 可选的服务端 ID 用于筛选工具

def get_server_tools(server_id: str) -> list[dict]

返回: 工具元数据字典列表

`get_all_tools()`

从所有服务器获取工具元数据。

def get_all_tools() -> list[dict]

`is_server_connected()`

检查服务器是否已连接。

def is_server_connected(server_id: str) -> bool

`get_connection_status()`

获取所有服务器的连接状态。

def get_connection_status() -> dict[str, bool]

返回: 服务器ID到连接状态的映射字典

`get_all_statuses()`

获取所有服务器的详细状态。

def get_all_statuses() -> dict[str, ConnectionStatus]

返回: 服务器ID到详细ConnectionStatus枚举的映射字典

`get_stats()`

获取客户端管理器统计信息。

def get_stats() -> dict

返回:

{
    'connected_servers': int,
    'total_tools': int,
    'server_status': dict[str, bool]
}

`add_event_listener()`

添加客户端事件监听器。

def add_event_listener(listener: Callable) -> None

`remove_event_listener()`

移除事件监听器。

def remove_event_listener(listener: Callable) -> bool

MCPClientTool

封装外部MCP工具以适配AG-Kit工具系统。

构造函数

MCPClientTool(
    mcp_client: Client,
    mcp_tool_metadata: dict,
    config: dict = None,
    adapter_config: dict = None
)

参数:

mcp_client: Client - MCP客户端实例
mcp_tool_metadata: dict - 来自MCP服务器的工具元数据
config: dict - 可选工具配置
adapter_config: dict - 可选适配器配置

工具元数据:

{
    'name': str,
    'description': str,      # 可选
    'inputSchema': dict      # 可选MCP JSON模式
}

工具配置:

{
    'name': str,       # 覆盖工具名称
    'timeout': int,    # 调用超时时间（毫秒）
    'retries': int     # 重试次数
}

适配器配置:

{
    'include_metadata': bool,              # 包含执行元数据
    'transform_input': Callable,           # MCP调用前转换输入
    'transform_output': Callable,          # MCP调用后转换输出
    'error_handler': Callable              # 自定义错误处理器
}

方法

`invoke()`

执行MCP工具（继承自BaseTool）。

async def invoke(input_data: Any, context: dict = None) -> ToolResult

`get_mcp_metadata()`

获取原始MCP工具元数据。

def get_mcp_metadata() -> dict

`is_connected()`

检查底层MCP客户端是否已连接。

def is_connected() -> bool

`update_config()`

更新工具配置。

def update_config(new_config: dict) -> None

`update_adapter_config()`

更新适配器配置。

def update_adapter_config(new_adapter_config: dict) -> None

MCPToolkit

用于管理外部服务器MCP工具的高级工具包。

构造函数

MCPToolkit(name: str = 'mcp-toolkit', description: str = None)

参数:

name: str - 工具包名称（默认：‘mcp-toolkit’）
description: str - 可选工具包描述

属性

`manager`

获取底层MCPClientManager实例。

@property
def manager() -> MCPClientManager

方法

`add_server()`

添加MCP服务器并加载其工具。

async def add_server(server_id: str, config: dict) -> None

`remove_server()`

移除服务器及其工具。

async def remove_server(server_id: str) -> None

`get_connected_servers()`

获取已连接服务器ID列表。

def get_connected_servers() -> list[str]

`get_server_tools()`

获取特定服务器的工具。

def get_server_tools(server_id: str) -> list[MCPClientTool]

`get_server_statuses()`

获取所有服务器的连接状态。

def get_server_statuses() -> dict[str, bool]

`refresh_tools()`

刷新特定服务器或所有服务器的工具。

async def refresh_tools(server_id: str = None) -> None

`destroy()`

清理所有资源并断开服务器连接。

async def destroy() -> None

`get_metadata()`

获取工具包元数据。

def get_metadata() -> dict

返回:

{
    'name': str,
    'description': str,
    'tool_count': int,
    'mcp': {
        'connected_servers': int,
        'total_tools': int,
        'server_status': dict[str, bool]
    }
}

实用函数

`pydantic_to_json_schema()`

将Pydantic模型转换为MCP JSON模式格式。

def pydantic_to_json_schema(model: type[BaseModel]) -> dict

参数:

model: type[BaseModel] - 待转换的Pydantic模型类

返回: MCP兼容的JSON模式字典

`convert_agkit_tool_to_mcp_metadata()`

将AG-Kit的BaseTool转换为MCP工具元数据。

def convert_agkit_tool_to_mcp_metadata(
    tool: BaseTool,
    config: dict = None,
    tool_config: dict = None
) -> dict

参数:

tool: BaseTool - AG-Kit工具实例
config: dict - 可选服务器配置
tool_config: dict - 可选工具特定配置

返回: MCP工具元数据字典

`create_agkit_mcp_server()`

创建并配置MCP服务器的实用函数。

async def create_agkit_mcp_server(
    config: dict,
    tools: list[BaseTool] = None
) -> AGKitMCPServer

`create_mcp_toolkit()`

创建带有预配置服务器的新MCP工具包。

async def create_mcp_toolkit(
    servers: list[dict],
    name: str = 'mcp-toolkit'
) -> MCPToolkit

参数:

servers: list[dict] - 服务器配置列表
name: str - 可选工具包名称

服务器配置:

{
    'id': str,
    'config': dict  # MCPClientConfig
}

服务器配置:

{
    'id': str,
    'config': dict  # MCPClientConfig
}

内存传输注册表

MemoryTransportRegistry

用于管理内存中MCP传输的单例注册表。

方法

`get_instance()`

获取单例实例。

@classmethod
def get_instance() -> MemoryTransportRegistry

`register_server()`

使用内存传输注册MCP服务器。

def register_server(memory_id: str, server: AGKitMCPServer) -> None

`unregister_server()`

注销服务器。

def unregister_server(memory_id: str) -> None

`get_server()`

获取已注册的服务器。

def get_server(memory_id: str) -> AGKitMCPServer | None

`create_transport_pair()`

创建一对关联的内存传输。

def create_transport_pair(memory_id: str) -> tuple[Any, Any]

返回: (server_transport, client_transport) 元组

`get_client_transport()`

获取用于连接的客户端传输。

def get_client_transport(memory_id: str) -> Any

`call_tool()`

通过内存传输直接调用工具。

async def call_tool(
    memory_id: str,
    tool_name: str,
    args: dict
) -> list

`list_tools()`

通过内存传输列出工具。

async def list_tools(memory_id: str) -> list[Tool]

类型定义

MCPEvent

MCP事件类。

class MCPEvent:
    type: MCPEventType
    data: dict

MCPEventType

MCP事件类型枚举。

class MCPEventType(str, Enum):
    CONNECTED = 'connected'
    DISCONNECTED = 'disconnected'
    TOOL_DISCOVERED = 'tool_discovered'
    TOOL_CALLED = 'tool_called'
    TOOL_RESULT = 'tool_result'
    ERROR = 'error'

ConnectionStatus

连接状态枚举。

class ConnectionStatus(str, Enum):
    CONNECTED = 'connected'
    DISCONNECTED = 'disconnected'
    CONNECTING = 'connecting'
    ERROR = 'error'

MCPServerConfig

服务器配置模型。

class MCPServerConfig(BaseModel):
    name: str
    version: str
    description: str = ''
    log_level: str = 'info'
    capabilities: dict = Field(default_factory=dict)

MCPClientConfig

客户端配置模型。

class MCPClientConfig(BaseModel):
    name: str
    version: str
    transport: MCPTransportConfig | None = None
    capabilities: dict = Field(default_factory=dict)

MCPToolMetadata

工具元数据模型。

class MCPToolMetadata(BaseModel):
    name: str
    description: str | None = None
    input_schema: dict | None = Field(default=None, alias='inputSchema')

MCPToolConfig

工具配置模型。

class MCPToolConfig(BaseModel):
    name: str | None = None
    description: str | None = None
    timeout: int = 30000
    retries: int = 0

MCPAdapterConfig

适配器配置模型。

class MCPAdapterConfig(BaseModel):
    include_metadata: bool = Field(default=True, alias='includeMetadata')
    transform_input: Callable | None = Field(default=None, alias='transformInput')
    transform_output: Callable | None = Field(default=None, alias='transformOutput')
    error_handler: Callable | None = Field(default=None, alias='errorHandler')

MCPToolRegistration

工具注册配置模型。

class MCPToolRegistration(BaseModel):
    name: str | None = None
    description: str | None = None
    name_prefix: str = Field(default='', alias='namePrefix')

传输配置模型

MCPStdioTransportConfig

class MCPStdioTransportConfig(BaseModel):
    type: Literal['stdio'] = 'stdio'
    command: str
    args: list[str] = Field(default_factory=list)
    timeout: int = 10000

MCPSSETransportConfig

class MCPSSETransportConfig(BaseModel):
    type: Literal['sse'] = 'sse'
    url: str
    timeout: int = 10000

MCPStreamableHTTPTransportConfig

class MCPStreamableHTTPTransportConfig(BaseModel):
    type: Literal['streamableHttp'] = 'streamableHttp'
    url: str
    timeout: int = 10000

MCPMemoryTransportConfig

class MCPMemoryTransportConfig(BaseModel):
    type: Literal['memory'] = 'memory'
    memory_id: str = Field(default='default', alias='memoryId')

错误处理

异常类

class MCPError(Exception):
    """MCP错误基类"""
    pass

class MCPConnectionError(MCPError):
    """连接相关错误"""
    pass

class MCPToolError(MCPError):
    """工具执行错误"""
    pass

class MCPTransportError(MCPError):
    """传输相关错误"""
    pass

模块导出

核心类

from ag_kit_py.tools.mcp import (
    AGKitMCPServer,
    MCPClientManager,
    MCPClientTool,
    MCPToolkit
)

工具函数

from ag_kit_py.tools.mcp import (
    pydantic_to_json_schema,
    convert_agkit_tool_to_mcp_metadata,
    create_agkit_mcp_server,
    create_mcp_toolkit
)

内存传输

from ag_kit_py.tools.mcp import (
    MemoryTransportRegistry
)

类型与模型

from ag_kit_py.tools.mcp.types import (
    MCPServerConfig,
    MCPClientConfig,
    MCPToolConfig,
    MCPAdapterConfig,
    MCPToolMetadata,
    MCPToolRegistration,
    MCPEvent,
    MCPEventType,
    ConnectionStatus,
    # ... 所有传输配置类型
)

使用模式

服务器模式

from ag_kit_py.tools.mcp import AGKitMCPServer

server = AGKitMCPServer(config)
server.register_tool(tool)
await server.run(transport_config)

客户端模式

from ag_kit_py.tools.mcp import MCPClientManager

manager = MCPClientManager()
await manager.add_server(server_id, client_config)
tools = manager.create_client_tools()

工具包模式

from ag_kit_py.tools.mcp import MCPToolkit

toolkit = MCPToolkit()
await toolkit.add_server(server_id, config)
tools = toolkit.get_tools()

本API参考提供了AG-Kit Python中MCP集成的所有公共接口和方法的完整文档。有关使用示例和实现模式，请参阅MCP集成指南和python-sdk/examples/中的示例文件。

TypeScript SDK

Python SDK

无头 UI 组件

CLI 工具

​MCP 应用程序接口参考

​核心类

​AGKitMCPServer

​构造函数

​方法

register_tool()

register_tools()

unregister_tool()

run()

stop()

call_tool()

list_tools()

is_server_running()

get_stats()

get_server_info()

get_config()

update_config()

get_connection_info()

get_transport_type()

get_memory_id()

add_event_listener()

remove_event_listener()

​MCPClientManager

​构造函数

​方法

add_server()

disconnect_server()

disconnect_all()

create_client_tools()

get_all_tools()

is_server_connected()

get_connection_status()

get_all_statuses()

get_stats()

add_event_listener()

remove_event_listener()

​MCPClientTool

​构造函数

​方法

invoke()

get_mcp_metadata()

is_connected()

update_config()

update_adapter_config()

​MCPToolkit

​构造函数

​属性

manager

​方法

add_server()

remove_server()

get_connected_servers()

get_server_tools()

get_server_statuses()

refresh_tools()

destroy()

get_metadata()

​实用函数

​pydantic_to_json_schema()

​convert_agkit_tool_to_mcp_metadata()

​create_agkit_mcp_server()

​create_mcp_toolkit()

​内存传输注册表

​MemoryTransportRegistry

​方法

get_instance()

register_server()

unregister_server()

get_server()

create_transport_pair()

get_client_transport()

call_tool()

list_tools()

​类型定义

​MCPEvent

​MCPEventType

MCP 应用程序接口参考

核心类

AGKitMCPServer

构造函数

方法

`register_tool()`

`register_tools()`

`unregister_tool()`

`run()`

`stop()`

`call_tool()`

`list_tools()`

`is_server_running()`

`get_stats()`

`get_server_info()`

`get_config()`

`update_config()`

`get_connection_info()`

`get_transport_type()`

`get_memory_id()`

`add_event_listener()`

`remove_event_listener()`

MCPClientManager

构造函数

方法

`add_server()`

`disconnect_server()`

`disconnect_all()`

`create_client_tools()`

`get_all_tools()`

`is_server_connected()`

`get_connection_status()`

`get_all_statuses()`

`get_stats()`

`add_event_listener()`

`remove_event_listener()`

MCPClientTool

构造函数

方法

`invoke()`

`get_mcp_metadata()`

`is_connected()`

`update_config()`

`update_adapter_config()`

MCPToolkit

构造函数

属性

`manager`

方法

`add_server()`

`remove_server()`

`get_connected_servers()`

`get_server_tools()`

`get_server_statuses()`

`refresh_tools()`

`destroy()`

`get_metadata()`

实用函数

`pydantic_to_json_schema()`

`convert_agkit_tool_to_mcp_metadata()`

`create_agkit_mcp_server()`

`create_mcp_toolkit()`

内存传输注册表

MemoryTransportRegistry

方法

`get_instance()`

`register_server()`

`unregister_server()`

`get_server()`

`create_transport_pair()`

`get_client_transport()`

`call_tool()`

`list_tools()`

类型定义

MCPEvent

MCPEventType

ConnectionStatus

MCPServerConfig

MCPClientConfig

MCPToolMetadata