Skip to main content

​
MCP API Reference

This document provides a complete API reference for AG-Kit Python’s Model Context Protocol (MCP) integration components.

​
Core Classes

​
AGKitMCPServer

Exposes AG-Kit tools as a standard MCP server.

​
Constructor

AGKitMCPServer(config: dict)
Parameters:
  • config: dict - Server configuration dictionary
Configuration Dictionary: 
{
    'name': str,                    # Server name (required)
    'version': str,                 # Server version (required)
    'description': str,             # Optional server description
    'log_level': str,               # Log level: 'info', 'debug' (default: 'info')
    'capabilities': dict,           # Optional MCP capabilities
    'error_handling': str,          # 'throw' or 'return_error' (default: 'return_error')
}

​
Methods

register_tool()
Register a single AG-Kit tool with the server. 
register_tool(tool: BaseTool, registration: dict = None) -> dict
Parameters:
  • tool: BaseTool - The AG-Kit tool to register
  • registration: dict - Optional tool registration configuration
Registration Configuration: 
{
    'name': str,           # Override tool name
    'description': str,    # Override tool description
    'name_prefix': str,    # Add prefix to tool name
}
Returns: dict - Metadata for the registered tool
register_tools()
Register multiple AG-Kit tools with shared configuration. 
register_tools(tools: list[BaseTool], registration: dict = None) -> list[dict]
Parameters:
  • tools: list[BaseTool] - List of AG-Kit tools to register
  • registration: dict - Optional shared configuration
Returns: list[dict] - List of metadata for registered tools
unregister_tool()
Remove a tool from the server. 
unregister_tool(name: str) -> bool
Parameters:
  • name: str - Name of the tool to remove
Returns: bool - True if tool was removed, False if not found
run()
Start the MCP server with specified transport. 
async def run(transport_config: dict) -> None
Parameters:
  • transport_config: dict - Transport configuration
Transport Configuration Types: 
# Stdio Transport
{
    'type': 'stdio'
}

# Memory Transport
{
    'type': 'memory',
    'memory_id': str  # Optional, defaults to 'default'
}

# SSE Transport
{
    'type': 'sse',
    'sse_setup': Callable  # Async setup function
}

# Streamable HTTP Transport
{
    'type': 'streamableHttp',
    'streamable_http_setup': Callable  # Async setup function
}
stop()
Stop the MCP server and cleanup resources. 
async def stop() -> None
call_tool()
Execute a registered tool directly. 
async def call_tool(name: str, args: dict) -> dict
Parameters:
  • name: str - Name of the tool to execute
  • args: dict - Tool arguments
Returns: dict - MCP-formatted tool result
list_tools()
Get list of all registered tools. 
def list_tools() -> list[dict]
Returns: List of tool metadata dictionaries
is_server_running()
Check if the server is currently running. 
def is_server_running() -> bool
Returns: bool - True if server is running
get_stats()
Get server statistics and information. 
def get_stats() -> dict
Returns: 
{
    'total_tools': int,
    'tool_names': list[str],
    'tool_types': dict[str, int],
    'is_running': bool,
    'server_info': {
        'name': str,
        'version': str,
        'description': str  # Optional
    }
}
get_server_info()
Get server information. 
def get_server_info() -> dict
Returns: 
{
    'name': str,
    'version': str,
    'description': str,  # Optional
    'tool_count': int,
    'tools': list[str]
}
get_config()
Get current server configuration. 
def get_config() -> dict
update_config()
Update server configuration. 
def update_config(new_config: dict) -> None
get_connection_info()
Get connection information. 
def get_connection_info() -> dict
Returns: 
{
    'is_running': bool,
    'transport_type': str | None,
    'memory_id': str | None
}
get_transport_type()
Get current transport type. 
def get_transport_type() -> str | None
get_memory_id()
Get memory transport ID if using memory transport. 
def get_memory_id() -> str | None
add_event_listener()
Add event listener for server events. 
def add_event_listener(listener: Callable) -> None
Parameters:
  • listener: Callable - Event listener function that takes an MCPEvent
remove_event_listener()
Remove event listener. 
def remove_event_listener(listener: Callable) -> bool
Returns: bool - True if listener was removed

​
MCPClientManager

Manages connections to external MCP servers.

​
Constructor

MCPClientManager()

​
Methods

add_server()
Add and connect to an MCP server. 
async def add_server(
    server_id: str,
    config: dict,
    connection_options: dict = None
) -> None
Parameters:
  • server_id: str - Unique identifier for the server
  • config: dict - Client configuration
  • connection_options: dict - Optional connection settings
Client Configuration: 
{
    'name': str,              # Client name (required)
    'version': str,           # Client version (required)
    'transport': dict,        # Transport config (required if client not provided)
    'client': Client,         # Pre-connected client (optional)
    'capabilities': dict,     # Optional capabilities
}
Connection Options: 
{
    'auto_reconnect': bool,         # Enable auto-reconnection (default: True)
    'reconnect_delay': int,         # Delay between reconnect attempts (default: 5000ms)
    'max_reconnect_attempts': int,  # Maximum reconnect attempts (default: 3)
    'heartbeat_interval': int,      # Heartbeat interval (default: 30000ms)
}
disconnect_server()
Disconnect from a specific server. 
async def disconnect_server(server_id: str) -> None
disconnect_all()
Disconnect from all servers. 
async def disconnect_all() -> None
create_client_tools()
Create AG-Kit tool wrappers for all available MCP tools. 
def create_client_tools(server_id: str = None) -> list[MCPClientTool]
Parameters:
  • server_id: str - Optional server ID to filter tools
Returns: list[MCPClientTool] - List of wrapped tools
create_client_tool()
Create a specific AG-Kit tool wrapper for an MCP tool. 
def create_client_tool(
    server_id: str,
    tool_name: str,
    agkit_tool_name: str = None
) -> MCPClientTool
Parameters:
  • server_id: str - Server ID containing the tool
  • tool_name: str - Name of the MCP tool
  • agkit_tool_name: str - Optional custom name for the AG-Kit tool
Returns: MCPClientTool - Wrapped tool instance
call_tool()
Call an MCP tool directly. 
async def call_tool(server_id: str, tool_name: str, args: Any) -> Any
get_server_tools()
Get tool metadata from a specific server. 
def get_server_tools(server_id: str) -> list[dict]
Returns: List of tool metadata dictionaries
get_all_tools()
Get tool metadata from all servers. 
def get_all_tools() -> list[dict]
is_server_connected()
Check if a server is connected. 
def is_server_connected(server_id: str) -> bool
get_connection_status()
Get connection status for all servers. 
def get_connection_status() -> dict[str, bool]
Returns: Dictionary mapping server IDs to connection status
get_all_statuses()
Get detailed status for all servers. 
def get_all_statuses() -> dict[str, ConnectionStatus]
Returns: Dictionary mapping server IDs to detailed ConnectionStatus enums
get_stats()
Get client manager statistics. 
def get_stats() -> dict
Returns: 
{
    'connected_servers': int,
    'total_tools': int,
    'server_status': dict[str, bool]
}
add_event_listener()
Add event listener for client events. 
def add_event_listener(listener: Callable) -> None
remove_event_listener()
Remove event listener. 
def remove_event_listener(listener: Callable) -> bool

​
MCPClientTool

Wraps an external MCP tool to work within AG-Kit’s tool system.

​
Constructor

MCPClientTool(
    mcp_client: Client,
    mcp_tool_metadata: dict,
    config: dict = None,
    adapter_config: dict = None
)
Parameters:
  • mcp_client: Client - MCP client instance
  • mcp_tool_metadata: dict - Tool metadata from MCP server
  • config: dict - Optional tool configuration
  • adapter_config: dict - Optional adapter configuration
Tool Metadata: 
{
    'name': str,
    'description': str,      # Optional
    'inputSchema': dict      # Optional MCP JSON schema
}
Tool Configuration: 
{
    'name': str,       # Override tool name
    'timeout': int,    # Call timeout in ms
    'retries': int     # Retry attempts
}
Adapter Configuration: 
{
    'include_metadata': bool,              # Include execution metadata
    'transform_input': Callable,           # Transform input before MCP call
    'transform_output': Callable,          # Transform output after MCP call
    'error_handler': Callable              # Custom error handler
}

​
Methods

invoke()
Execute the MCP tool (inherited from BaseTool). 
async def invoke(input_data: Any, context: dict = None) -> ToolResult
get_mcp_metadata()
Get the original MCP tool metadata. 
def get_mcp_metadata() -> dict
is_connected()
Check if the underlying MCP client is connected. 
def is_connected() -> bool
update_config()
Update tool configuration. 
def update_config(new_config: dict) -> None
update_adapter_config()
Update adapter configuration. 
def update_adapter_config(new_adapter_config: dict) -> None

​
MCPToolkit

High-level toolkit for managing MCP tools from external servers.

​
Constructor

MCPToolkit(name: str = 'mcp-toolkit', description: str = None)
Parameters:
  • name: str - Toolkit name (default: ‘mcp-toolkit’)
  • description: str - Optional toolkit description

​
Properties

manager
Get the underlying MCPClientManager. 
@property
def manager() -> MCPClientManager

​
Methods

add_server()
Add an MCP server and load its tools. 
async def add_server(server_id: str, config: dict) -> None
remove_server()
Remove a server and its tools. 
async def remove_server(server_id: str) -> None
get_connected_servers()
Get list of connected server IDs. 
def get_connected_servers() -> list[str]
get_server_tools()
Get tools from a specific server. 
def get_server_tools(server_id: str) -> list[MCPClientTool]
get_server_statuses()
Get connection status for all servers. 
def get_server_statuses() -> dict[str, bool]
refresh_tools()
Refresh tools from a specific server or all servers. 
async def refresh_tools(server_id: str = None) -> None
destroy()
Cleanup all resources and disconnect from servers. 
async def destroy() -> None
get_metadata()
Get toolkit metadata. 
def get_metadata() -> dict
Returns: 
{
    'name': str,
    'description': str,
    'tool_count': int,
    'mcp': {
        'connected_servers': int,
        'total_tools': int,
        'server_status': dict[str, bool]
    }
}

​
Utility Functions

​
pydantic_to_json_schema()

Convert a Pydantic model to MCP JSON schema format. 
def pydantic_to_json_schema(model: type[BaseModel]) -> dict
Parameters:
  • model: type[BaseModel] - Pydantic model class to convert
Returns: MCP-compatible JSON schema dictionary

​
convert_agkit_tool_to_mcp_metadata()

Convert AG-Kit BaseTool to MCP tool metadata. 
def convert_agkit_tool_to_mcp_metadata(
    tool: BaseTool,
    config: dict = None,
    tool_config: dict = None
) -> dict
Parameters:
  • tool: BaseTool - AG-Kit tool instance
  • config: dict - Optional server config
  • tool_config: dict - Optional tool-specific config
Returns: MCP tool metadata dictionary

​
create_agkit_mcp_server()

Utility function to create and configure an MCP server. 
async def create_agkit_mcp_server(
    config: dict,
    tools: list[BaseTool] = None
) -> AGKitMCPServer

​
create_mcp_toolkit()

Create a new MCP toolkit with pre-configured servers. 
async def create_mcp_toolkit(
    servers: list[dict],
    name: str = 'mcp-toolkit'
) -> MCPToolkit
Parameters:
  • servers: list[dict] - List of server configurations
  • name: str - Optional toolkit name
Server Configuration: 
{
    'id': str,
    'config': dict  # MCPClientConfig
}

​
Memory Transport Registry

​
MemoryTransportRegistry

Singleton registry for managing in-memory MCP transports.

​
Methods

get_instance()
Get the singleton instance. 
@classmethod
def get_instance() -> MemoryTransportRegistry
register_server()
Register an MCP server with memory transport. 
def register_server(memory_id: str, server: AGKitMCPServer) -> None
unregister_server()
Unregister a server. 
def unregister_server(memory_id: str) -> None
get_server()
Get a registered server. 
def get_server(memory_id: str) -> AGKitMCPServer | None
create_transport_pair()
Create a linked pair of memory transports. 
def create_transport_pair(memory_id: str) -> tuple[Any, Any]
Returns: Tuple of (server_transport, client_transport)
get_client_transport()
Get client transport for connections. 
def get_client_transport(memory_id: str) -> Any
call_tool()
Direct tool call via memory transport. 
async def call_tool(
    memory_id: str,
    tool_name: str,
    args: dict
) -> list
list_tools()
List tools via memory transport. 
async def list_tools(memory_id: str) -> list[Tool]

​
Type Definitions

​
MCPEvent

Event class for MCP events. 
class MCPEvent:
    type: MCPEventType
    data: dict

​
MCPEventType

Enumeration of MCP event types. 
class MCPEventType(str, Enum):
    CONNECTED = 'connected'
    DISCONNECTED = 'disconnected'
    TOOL_DISCOVERED = 'tool_discovered'
    TOOL_CALLED = 'tool_called'
    TOOL_RESULT = 'tool_result'
    ERROR = 'error'

​
ConnectionStatus

Enumeration of connection statuses. 
class ConnectionStatus(str, Enum):
    CONNECTED = 'connected'
    DISCONNECTED = 'disconnected'
    CONNECTING = 'connecting'
    ERROR = 'error'

​
MCPServerConfig

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

​
MCPClientConfig

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

​
MCPToolMetadata

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

​
MCPToolConfig

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

​
MCPAdapterConfig

Adapter configuration model. 
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

Tool registration configuration model. 
class MCPToolRegistration(BaseModel):
    name: str | None = None
    description: str | None = None
    name_prefix: str = Field(default='', alias='namePrefix')

​
Transport Configuration Models

​
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')

​
Error Handling

​
Exception Classes

class MCPError(Exception):
    """Base exception for MCP errors."""
    pass

class MCPConnectionError(MCPError):
    """Connection-related errors."""
    pass

class MCPToolError(MCPError):
    """Tool execution errors."""
    pass

class MCPTransportError(MCPError):
    """Transport-related errors."""
    pass

​
Module Exports

​
Core Classes

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

​
Utility Functions

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

​
Memory Transport

from ag_kit_py.tools.mcp import (
    MemoryTransportRegistry
)

​
Types and Models

from ag_kit_py.tools.mcp.types import (
    MCPServerConfig,
    MCPClientConfig,
    MCPToolConfig,
    MCPAdapterConfig,
    MCPToolMetadata,
    MCPToolRegistration,
    MCPEvent,
    MCPEventType,
    ConnectionStatus,
    # ... all transport config types
)

​
Usage Patterns

​
Server Pattern

from ag_kit_py.tools.mcp import AGKitMCPServer

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

​
Client Pattern

from ag_kit_py.tools.mcp import MCPClientManager

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

​
Toolkit Pattern

from ag_kit_py.tools.mcp import MCPToolkit

toolkit = MCPToolkit()
await toolkit.add_server(server_id, config)
tools = toolkit.get_tools()
This API reference provides complete documentation for all public interfaces and methods in AG-Kit Python’s MCP integration. For usage examples and implementation patterns, see the MCP Integration Guide and the example files in python-sdk/examples/.