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[source]¶

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()[source]¶

Return type:: None

cleanup()[source]¶

Return type:: None

list_tools()[source]¶

Return type:: Sequence[Tool]

list_resources()[source]¶

Return type:: Sequence[Resource]

list_resource_contents(uri)[source]¶

Parameters:: uri (str)
Return type:: Sequence[TextResourceContents | BlobResourceContents]

call_tool(tool_name, arguments)[source]¶

Parameters:

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

Return type:

CallToolResult

abstractmethod async connect_async()[source]¶

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()[source]¶

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()[source]¶

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()[source]¶

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)[source]¶

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)[source]¶

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¶

Streamable HTTP implementation of the Model Context Protocol (MCP) server client.

This module provides an HTTP client implementation for interacting with MCP servers using the Streamable HTTP transport as defined in the MCP 2025-03-26 specification. It enables connection management, tool discovery, resource querying, and tool execution through a standardized MCP endpoint.

The implementation follows the MCPServerProtocol interface and uses httpx for asynchronous HTTP communication.

class StreamableHTTPMCPServer(*, server_name, server_url, mcp_endpoint='/mcp', headers=<factory>, timeout_s=100.0, session_manager=<factory>)[source]¶

Bases: MCPServerProtocol

Streamable HTTP implementation of the MCP (Model Context Protocol) server client.

This class provides a client implementation for interacting with remote MCP servers over HTTP using the Streamable HTTP transport (MCP 2025-03-26 spec). It supports both regular and streaming responses, session management, and handles connection management, tool discovery, resource management, and tool execution.

Parameters:

server_name (str)
server_url (str)
mcp_endpoint (str | Callable[[...], str])
headers (MutableMapping[str, str])
timeout_s (float)
session_manager (SessionManager)

server_name¶

A human-readable name for the server

Type:: str

server_url¶

The base URL of the HTTP server

Type:: AnyUrl

mcp_endpoint¶

The endpoint path for MCP requests (e.g., “/mcp”)

Type:: str

headers¶

HTTP headers to include with each request

Type:: MutableMapping[str, str]

timeout_s¶

Request timeout in seconds

Type:: float

session_manager¶

Manager for storing session information

Type:: SessionManager

Usage:: server = StreamableHTTPMCPServer(server_name=”Example MCP”, server_url=”http://example.com”, mcp_endpoint=”/mcp”) await server.connect() tools = await server.list_tools() result = await server.call_tool(“tool_name”, {“param”: “value”}) await server.cleanup()

server_name: str¶

server_url: str¶

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

headers: MutableMapping[str, str]¶

timeout_s: float¶

session_manager: SessionManager¶

property name: str¶

Get a readable name for the server.

Returns:: The human-readable server name
Return type:: str

async connect_async()[source]¶

Connect to the HTTP MCP server and initialize the MCP protocol.

Establishes an HTTP client connection to the server and performs the initialization handshake as defined in the MCP specification.

Raises:: ConnectionError – If the connection to the server cannot be established
Return type:: None

async cleanup_async()[source]¶

Clean up the server connection.

Closes the HTTP client connection if it exists. If a session ID was established, attempts to terminate the session with a DELETE request.

Returns:: None
Return type:: None

async list_tools_async()[source]¶

List the tools available on the server.

Returns:

A list of Tool objects available on the server

Return type:

Sequence[Tool]

Raises:

ConnectionError – If the server is not connected
httpx.RequestError – If there’s an error during the HTTP request

async list_resources_async()[source]¶

List the resources available on the server.

Returns:

A list of Resource objects available on the server

Return type:

Sequence[Resource]

Raises:

ConnectionError – If the server is not connected
httpx.RequestError – If there’s an error during the HTTP request

async list_resource_contents_async(uri)[source]¶

List contents of a specific resource.

Parameters:

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

Returns:

A list of resource contents

Return type:

Sequence[TextResourceContents | BlobResourceContents]

Raises:

ConnectionError – If the server is not connected
httpx.RequestError – If there’s an error during the HTTP request

async call_tool_async(tool_name, arguments)[source]¶

Invoke a tool on the server.

Parameters:

tool_name (str) – The name of the tool to call
arguments (MutableMapping[str, object] | None) – The arguments to pass to the tool

Returns:

The result of the tool invocation

Return type:

CallToolResult

Raises:

ConnectionError – If the server is not connected
httpx.RequestError – If there’s an error during the HTTP request

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¶

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

This module provides a client implementation for interacting with MCP servers over standard input/output streams. It enables connection management, tool discovery, resource querying, and tool execution through stdin/stdout communication.

The implementation follows the MCPServerProtocol interface and uses asyncio streams for communication.

class StdioMCPServer(*, server_name, command, server_env=<factory>, working_dir=None, request_timeout_s=100.0)[source]¶

Bases: MCPServerProtocol

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

This class provides a client implementation for interacting with MCP servers over standard input/output streams. It handles connection management, tool discovery, resource management, and tool invocation through stdin/stdout communication.

Parameters:

server_name (str)
command (str | Callable[[...], str])
server_env (MutableMapping[str, str])
working_dir (str | None)
request_timeout_s (float)

server_name¶

A human-readable name for the server

Type:: str

command¶

The command to run the server executable

Type:: str

server_env¶

Environment variables to pass to the server process

Type:: MutableMapping[str, str]

working_dir¶

Working directory for the server process

Type:: str

Usage:

server = StdioMCPServer(: server_name=”Example MCP”, command=”path/to/server”, server_env={“DEBUG”: “1”}, working_dir=”/path/to/working/dir”

) await server.connect() tools = await server.list_tools() result = await server.call_tool(“tool_name”, {“param”: “value”}) await server.cleanup()

server_name: str¶

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

server_env: MutableMapping[str, str]¶

working_dir: str | None¶

request_timeout_s: float¶

async connect_async()[source]¶

Connect to the MCP server over stdin/stdout.

Launches the server process and sets up communication channels. Initializes the MCP protocol.

Raises:

ConnectionError – If the server process cannot be started
TimeoutError – If the initialization takes too long

Return type:

None

property name: str¶

Get a readable name for the server.

Returns:: The human-readable server name
Return type:: str

async cleanup_async()[source]¶

Clean up the server connection.

Terminates the server process and cleans up resources.

Return type:: None

async list_tools_async()[source]¶

List the tools available on the server.

Returns:: A list of Tool objects available on the server
Return type:: Sequence[Tool]
Raises:: ConnectionError – If the server is not connected

async list_resources_async()[source]¶

List the resources available on the server.

Returns:: A list of Resource objects available on the server
Return type:: Sequence[Resource]
Raises:: ConnectionError – If the server is not connected

async list_resource_contents_async(uri)[source]¶

List contents of a specific resource.

Parameters:: uri (str) – The URI of the resource to retrieve contents for
Returns:: A list of resource contents
Return type:: Sequence[TextResourceContents | BlobResourceContents]
Raises:: ConnectionError – If the server is not connected

async call_tool_async(tool_name, arguments)[source]¶

Invoke a tool on the server.

Parameters:

tool_name (str) – The name of the tool to call
arguments (MutableMapping[str, object] | None) – The arguments to pass to the tool

Returns:

The result of the tool invocation

Return type:

CallToolResult

Raises:

ConnectionError – If the server is not connected

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[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Delete session information for a specific server.

Parameters:: server_key (str) – A unique identifier for the server connection
Return type:: None

abstractmethod async close()[source]¶

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[source]¶

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__()[source]¶

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

Return type:: None

async get_session(server_key)[source]¶

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)[source]¶

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)[source]¶

Delete session information for a specific server.

Parameters:: server_key (str) – A unique identifier for the server connection
Return type:: None

async close()[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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()[source]¶

Close the Redis connection.

Return type:: None