MCP API

This section documents the Model Context Protocol (MCP) implementation, including server interfaces and session management.

agentle.mcp	Model Context Protocol (MCP) Package
agentle.mcp.servers	MCP Servers Package
agentle.mcp.session_management	Session management for MCP servers.

MCP Server Protocol

MCP Server Protocol Module

This module defines the abstract base class for Model Context Protocol servers. It provides a standardized interface for different server implementations to connect to external resources, list available tools, and invoke tools.

class MCPServerProtocol

Bases: BaseModel, ABC

Abstract base class defining the protocol for MCP servers.

This class establishes the common interface that all MCP server implementations must adhere to, including connection management, tool discovery, resource listing, and tool invocation.

Implementing classes must provide concrete implementations for all abstract methods defined in this interface.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

abstract property name: str

Get a readable name for the server.

Returns:

A human-readable name identifying the server.

Return type:

str

connect()

Return type:

None

cleanup()

Return type:

None

list_tools()

Return type:

Sequence[Tool]

list_resources()

Return type:

Sequence[Resource]

list_resource_contents(uri)

Parameters:

uri (str)

Return type:

Sequence[TextResourceContents | BlobResourceContents]

call_tool(tool_name, arguments)

Parameters:

• tool_name (str)

• arguments (MutableMapping[str, object] | None)

Return type:

CallToolResult

abstractmethod async connect_async()

Connect to the MCP server.

Establishes a connection to the server, which might involve spawning a subprocess, opening a network connection, or other initialization steps. The server is expected to remain connected until cleanup() is called.

Returns:

None

Raises:

ConnectionError – If connection cannot be established.

Return type:

None

abstractmethod async cleanup_async()

Clean up the server connection.

Performs necessary cleanup operations such as closing a subprocess, terminating a network connection, or releasing other resources.

Returns:

None

Return type:

None

abstractmethod async list_tools_async()

List the tools available on the server.

Retrieves a list of tools that are available for use through this server.

Returns:

A sequence of Tool objects describing the available tools.

Return type:

Sequence[Tool]

Raises:

ConnectionError – If the server is not connected.

abstractmethod async list_resources_async()

List the resources available on the server.

Retrieves a list of resources that are available through this server.

Returns:

A sequence of Resource objects describing the available resources.

Return type:

Sequence[Resource]

Raises:

ConnectionError – If the server is not connected.

abstractmethod async list_resource_contents_async(uri)

List the contents of a specific resource.

Retrieves the contents of a resource identified by its URI.

Parameters:

uri (str) – The URI of the resource to retrieve contents for.

Returns:

A sequence of resource content objects, which can be either text or binary data.

Return type:

Sequence[TextResourceContents | BlobResourceContents]

Raises:

• ConnectionError – If the server is not connected.

• ValueError – If the URI is invalid or the resource does not exist.

abstractmethod async call_tool_async(tool_name, arguments)

Invoke a tool on the server.

Calls a specified tool with the provided arguments and returns the result.

Parameters:

• tool_name (str) – The name of the tool to call.

• arguments (MutableMapping[str, object] | None) – Arguments to pass to the tool, or None if no arguments.

Returns:

The result of the tool invocation.

Return type:

CallToolResult

Raises:

• ConnectionError – If the server is not connected.

• ValueError – If the tool does not exist or the arguments are invalid.

Streamable HTTP MCP Server

Production-ready fixes for StreamableHTTPMCPServer using aiohttp

This version fixes the JSON-RPC 2.0 compliance issues and protocol version problems.

Key fixes: 1. Updated protocol version to “2025-06-18” 2. Fixed notification method name to “notifications/initialized” 3. Ensured proper endpoint handling to avoid redirects 4. Improved parameter handling for requests 5. Better error handling and logging

class StreamableHTTPMCPServer(*, server_name, server_url, mcp_endpoint='/mcp/', headers=<factory>, timeout_s=100.0, session_manager=<factory>, max_retries=3, retry_delay_s=1.0, connection_pool_limits=<factory>)

Bases: MCPServerProtocol

Production-ready Streamable HTTP implementation of MCP server client using aiohttp.

This version uses aiohttp.ClientSession for improved performance and follows the MCP 2025-06-18 protocol specification.

Parameters:

• server_name (str)

• server_url (str)

• mcp_endpoint (str | Callable[[...], str])

• headers (MutableMapping[str, str])

• timeout_s (float)

• session_manager (SessionManager)

• max_retries (int)

• retry_delay_s (float)

• connection_pool_limits (Dict[str, int])

server_name: str

server_url: str

mcp_endpoint: str | Callable[..., str]

headers: MutableMapping[str, str]

timeout_s: float

session_manager: SessionManager

max_retries: int

retry_delay_s: float

connection_pool_limits: Dict[str, int]

property name: str

Get a readable name for the server.

Returns:

A human-readable name identifying the server.

Return type:

str

async connect_async()

Connect with proper session restoration and validation.

Return type:

None

async cleanup_async()

Clean up resources with proper session termination.

Return type:

None

async list_tools_async()

List tools with error handling.

Return type:

Sequence[Tool]

async list_resources_async()

List resources with error handling.

Return type:

Sequence[Resource]

async list_resource_contents_async(uri)

List resource contents with error handling.

Parameters:

uri (str)

Return type:

Sequence[TextResourceContents | BlobResourceContents]

async call_tool_async(tool_name, arguments)

Call tool with error handling.

Parameters:

• tool_name (str)

• arguments (MutableMapping[str, object] | None)

Return type:

CallToolResult

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

Return type:

None

Stdio MCP Server

Production-ready Stdio implementation of the Model Context Protocol (MCP) server client.

FIXES APPLIED: 1. Updated protocol version to 2025-06-18 (current version) 2. Fixed client capabilities structure (roots, sampling, elicitation) 3. Added progress tracking support with callbacks 4. Added cancellation support (notifications/cancelled) 5. Added ping/pong functionality for health monitoring 6. Improved response validation and error handling 7. Fixed connection state race conditions 8. Added configurable logging levels 9. Enhanced error messages with more context 10. Added proper request ID management with UUIDs

ENHANCEMENTS: 1. Progress token support for long-running operations 2. Connection state machine for better state management 3. Retry logic capabilities 4. Better resource cleanup and error recovery 5. Comprehensive protocol compliance with MCP 2025-06-18

class ConnectionState(*values)

Bases: Enum

Connection state enumeration for better state management.

DISCONNECTED = 'disconnected'

CONNECTING = 'connecting'

INITIALIZING = 'initializing'

CONNECTED = 'connected'

DISCONNECTING = 'disconnecting'

FAILED = 'failed'

class StdioMCPServer(*, server_name, command, server_env=<factory>, working_dir=None, request_timeout_s=30.0, startup_timeout_s=10.0, shutdown_timeout_s=5.0, log_level='INFO', health_check_interval_s=5.0, process_startup_delay_s=0.1, max_connection_retries=3, retry_delay_s=1.0)

Bases: MCPServerProtocol

Production-ready Stdio implementation of the MCP (Model Context Protocol) server client.

FIXED AND ENHANCED VERSION with full MCP 2025-06-18 protocol compliance.

This class provides a client implementation for interacting with MCP servers over standard input/output streams. The server is launched as a subprocess and communication happens through stdin/stdout pipes.

Key Features: - Full MCP protocol compliance (2025-06-18) - Progress tracking with callbacks - Request cancellation support - Ping/pong health monitoring - Robust process lifecycle management - Connection state machine - Configurable logging levels - Graceful error handling and recovery - Resource leak prevention

Parameters:

• server_name (str)

• command (str | Callable[[...], str])

• server_env (MutableMapping[str, str])

• working_dir (str | None)

• request_timeout_s (float)

• startup_timeout_s (float)

• shutdown_timeout_s (float)

• log_level (str)

• health_check_interval_s (float)

• process_startup_delay_s (float)

• max_connection_retries (int)

• retry_delay_s (float)

server_name

A human-readable name for the server

Type:

str

command

The command to launch the MCP server subprocess

Type:

str

server_env

Environment variables for the server process

Type:

MutableMapping[str, str]

working_dir

Working directory for the server process

Type:

str

request_timeout_s

Timeout for individual requests

Type:

float

startup_timeout_s

Timeout for server startup

Type:

float

shutdown_timeout_s

Timeout for graceful shutdown

Type:

float

log_level

Logging level for the client

Type:

str

health_check_interval_s

Interval for health monitoring

Type:

float

process_startup_delay_s

Delay after process startup

Type:

float

Usage:

server = StdioMCPServer(

server_name=”OpenMemory MCP”, command=”npx openmemory”, server_env={“OPENMEMORY_API_KEY”: “your-key”}, log_level=”INFO”

)

try:

await server.connect() tools = await server.list_tools()

# With progress tracking def on_progress(current: int, total: int | None, message: str | None):

print(f”Progress: {current}/{total or ‘?’} - {message or ‘’}”)

result = await server.call_tool_async(“search”, {“query”: “test”}, on_progress)

finally:

await server.cleanup()

server_name: str

command: str | Callable[..., str]

server_env: MutableMapping[str, str]

working_dir: str | None

request_timeout_s: float

startup_timeout_s: float

shutdown_timeout_s: float

log_level: str

health_check_interval_s: float

process_startup_delay_s: float

max_connection_retries: int

retry_delay_s: float

property name: str

Get a readable name for the server.

property connection_state: ConnectionState

Get the current connection state.

async connect_async()

Connect to the MCP server over stdin/stdout with full protocol compliance.

Return type:

None

async cancel_request(request_id, reason='Request cancelled')

Send a cancellation notification for an in-progress request.

Parameters:

• request_id (str)

• reason (str)

Return type:

None

async ping_async(timeout=5.0)

Test server responsiveness with a ping request.

Parameters:

timeout (float)

Return type:

bool

ping(timeout=5.0)

Synchronous wrapper for ping_async.

Parameters:

timeout (float)

Return type:

bool

async cleanup_async()

Clean up the server connection with proper resource management.

Return type:

None

async list_tools_async()

List the tools available on the server.

Return type:

Sequence[Tool]

async list_resources_async()

List the resources available on the server.

Return type:

Sequence[Resource]

async list_resource_contents_async(uri)

List contents of a specific resource.

Parameters:

uri (str)

Return type:

Sequence[TextResourceContents | BlobResourceContents]

async call_tool_async(tool_name, arguments, progress_callback=None)

Invoke a tool on the server with optional progress tracking.

Parameters:

• tool_name (str)

• arguments (MutableMapping[str, object] | None)

• progress_callback (ProgressCallback | None)

Return type:

CallToolResult

get_process_info()

Get information about the server process for debugging.

Return type:

dict[str, Any]

async wait_for_connection(timeout=30.0)

Wait for the server to be fully connected.

Parameters:

timeout (float)

Return type:

bool

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

Return type:

None

Session Management

Abstract base class defining the interface for MCP session management.

This module provides the interface that all session management implementations must follow, ensuring consistent behavior across different storage backends.

class SessionManager

Bases: ABC

Abstract base class for MCP session management.

This class defines the interface that all session management implementations must implement to handle MCP session data across requests and potentially across different processes.

abstractmethod async get_session(server_key)

Retrieve session information for a specific server.

Parameters:

server_key (str) – A unique identifier for the server connection

Returns:

The session data if it exists, None otherwise

Return type:

Optional[Dict[str, Any]]

abstractmethod async store_session(server_key, session_data)

Store session information for a specific server.

Parameters:

• server_key (str) – A unique identifier for the server connection

• session_data (Dict[str, Any]) – The session data to store

Return type:

None

abstractmethod async delete_session(server_key)

Delete session information for a specific server.

Parameters:

server_key (str) – A unique identifier for the server connection

Return type:

None

abstractmethod async close()

Close any resources used by the session manager.

Return type:

None

In-memory implementation of the MCP session manager.

This module provides a thread-safe in-memory session manager implementation suitable for use in single-process applications.

class InMemorySessionManager

Bases: SessionManager

Thread-safe in-memory implementation of the SessionManager interface.

This implementation stores session data in an in-memory dictionary and uses a threading lock to ensure thread safety. It is suitable for single-process applications but will not share session data across multiple processes or workers.

__init__()

Initialize the in-memory session manager with an empty store.

Return type:

None

async get_session(server_key)

Retrieve session information for a specific server.

Parameters:

server_key (str) – A unique identifier for the server connection

Returns:

The session data if it exists, None otherwise

Return type:

Optional[Dict[str, Any]]

async store_session(server_key, session_data)

Store session information for a specific server.

Parameters:

• server_key (str) – A unique identifier for the server connection

• session_data (Dict[str, Any]) – The session data to store

Return type:

None

async delete_session(server_key)

Delete session information for a specific server.

Parameters:

server_key (str) – A unique identifier for the server connection

Return type:

None

async close()

Close any resources used by the session manager.

For the in-memory implementation, this is a no-op as there are no external resources to close.

Return type:

None

Redis-based implementation of the MCP session manager.

This module provides a Redis-backed session manager implementation suitable for use in multi-process applications and production environments.

class RedisSessionManager(redis_url='redis://localhost:6379/0', key_prefix='mcp_session:', expiration_seconds=3600)

Bases: SessionManager

Redis-backed implementation of the SessionManager interface.

This implementation stores session data in Redis, making it suitable for use in multi-process applications and production environments where session data needs to be shared across multiple workers.

Parameters:

• redis_url (str)

• key_prefix (str)

• expiration_seconds (int)

__init__(redis_url='redis://localhost:6379/0', key_prefix='mcp_session:', expiration_seconds=3600)

Initialize the Redis session manager.

Parameters:

• redis_url (str) – Redis connection URL

• key_prefix (str) – Prefix for Redis keys to avoid collisions

• expiration_seconds (int) – Time in seconds before sessions expire

Return type:

None

async get_session(server_key)

Retrieve session information for a specific server from Redis.

Parameters:

server_key (str) – A unique identifier for the server connection

Returns:

The session data if it exists, None otherwise

Return type:

Optional[Dict[str, Any]]

async store_session(server_key, session_data)

Store session information for a specific server in Redis.

Parameters:

• server_key (str) – A unique identifier for the server connection

• session_data (Dict[str, Any]) – The session data to store

Return type:

None

async delete_session(server_key)

Delete session information for a specific server from Redis.

Parameters:

server_key (str) – A unique identifier for the server connection

Return type:

None

async close()

Close the Redis connection.

Return type:

None