API Reference

Protocols

MCPClientTransport

source

Protocol for outbound MCP transports.

A transport is responsible for sending JSON-RPC requests to an external MCP server and returning the parsed JSON-RPC response. Each call maps to one logical request-response pair; the transport handles any framing details (length-delimited lines, HTTP bodies, etc.).

connect

async def connect() -> None

source

Open the underlying connection (spawn process, open socket, etc.).

disconnect

async def disconnect() -> None

source

Close the underlying connection and free resources.

send

async def send(message: dict[str, Any]) -> None

source

Write a JSON-RPC message to the server.

Parameters

Parameter	Type	Description
`message`	dict[str, Any]	Fully-formed JSON-RPC request or notification dict.

receive

async def receive() -> dict[str, Any]

source

Read and return the next JSON-RPC message from the server.

Returns

Type	Description
dict[str, Any]	Parsed JSON-RPC response dict.

Raises

Exception	Description
MCPTransportError	On I/O failure or malformed data.

---

MCPPromptHandlerProtocol

source

Protocol for handling MCP prompt-related methods.

Handles prompts/list and prompts/get methods.

list_prompts

async def list_prompts() -> list[dict[str, Any]]

source

Handle prompts/list method.

Returns list of available prompt templates.

get_prompt

async def get_prompt(
    name: str,
    arguments: dict[str, Any] | None = None
) -> dict[str, Any]

source

Handle prompts/get method.

Returns a prompt with arguments filled in.

---

MCPPromptProviderProtocol

source

Protocol for providing prompt templates to the MCP server.

list_prompts

async def list_prompts() -> list[dict[str, Any]]

source

List available prompt templates.

get_prompt

async def get_prompt(
    name: str,
    arguments: dict[str, Any] | None = None
) -> dict[str, Any]

source

Get a prompt template with arguments filled.

---

MCPResourceHandlerProtocol

source

Protocol for handling MCP resource-related methods.

Handles resources/list, resources/read, and resources/templates/list.

list_resources

async def list_resources() -> list[dict[str, Any]]

source

Handle resources/list method.

Returns list of available resources.

read_resource

async def read_resource(uri: str) -> dict[str, Any]

source

Handle resources/read method.

Returns content of a specific resource.

list_templates

async def list_templates() -> list[dict[str, Any]]

source

Handle resources/templates/list method.

Returns list of URI templates for resources.

---

MCPResourceProviderProtocol

source

Protocol for providing resources to the MCP server.

Resources are data that AI clients can browse and read — database records, files, configuration, etc.

list_resources

async def list_resources() -> list[dict[str, Any]]

source

List available resources in MCP format.

read_resource

async def read_resource(uri: str) -> dict[str, Any]

source

Read a resource by URI.

---

MCPServerProtocol

source

Protocol for the MCP server.

The server routes JSON-RPC messages to handlers and returns JSON-RPC responses.

handle_message

async def handle_message(message: dict[str, Any]) -> dict[str, Any] | None

source

Handle a JSON-RPC message and return the response.

Returns None for notifications (no response expected).

---

MCPToolHandlerProtocol

source

Protocol for handling MCP tool-related methods.

Handles tools/list and tools/call methods.

list_tools

async def list_tools() -> list[dict[str, Any]]

source

Handle tools/list method.

Returns list of available tools with their definitions.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any]
) -> dict[str, Any]

source

Handle tools/call method.

Executes a tool and returns the result.

---

MCPToolProviderProtocol

source

Protocol for providing tools to the MCP server.

Satisfied by:

• ToolRegistryAdapter (bridges lexigram-agents ToolRegistry)
• Any custom tool provider

list_tools

async def list_tools() -> list[dict[str, Any]]

source

List all available tools in MCP format.

Returns list of dicts with name, description, inputSchema keys.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any]
) -> dict[str, Any]

source

Execute a tool by name with given arguments.

---

MCPTransportProtocol

source

Protocol for MCP transport implementations.

A transport handles the I/O layer — reading requests and writing responses. The MCP server is transport-agnostic.

start

async def start() -> None

source

Start the transport.

stop

async def stop() -> None

source

Stop the transport.

---

Classes

AbstractTransport

source

Abstract base class for MCP transport implementations.

A transport handles the I/O layer - reading requests and writing responses. The MCP server is transport-agnostic.

start

async def start() -> None

source

Start the transport.

stop

async def stop() -> None

source

Stop the transport.

send

async def send(message: dict[str, Any]) -> None

source

Send a message through the transport.

Parameters

Parameter	Type	Description
`message`	dict[str, Any]	JSON-RPC message to send.

receive

async def receive() -> dict[str, Any] | None

source

Receive a message from the transport.

Returns

Type	Description
dict[str, Any] | None	Parsed JSON-RPC message, or None if no message available.

---

ControllerPromptProvider

source

MCPPromptProviderProtocol backed by a list of MCPController instances.

Aggregates all @prompt-decorated methods and dispatches get_prompt calls to the matching handler.

__init__

def __init__(controller_instances: list[MCPController]) -> None

source

Initialize with a list of already-constructed controller instances.

Parameters

Parameter	Type	Description
`controller_instances`	list[MCPController]	Instantiated MCPController objects to aggregate.

list_prompts

async def list_prompts() -> list[dict[str, Any]]

source

List all prompt templates from registered controllers in MCP format.

Returns

Type	Description
list[dict[str, Any]]	MCP prompt definitions with ``name``, ``description``, ``arguments``.

get_prompt

async def get_prompt(
    name: str,
    arguments: dict[str, Any] | None = None
) -> dict[str, Any]

source

Get a prompt template by name with arguments filled in.

Parameters

Parameter	Type	Description
`name`	str	Prompt name.
`arguments`	dict[str, Any] | None	Optional arguments dict from the MCP client.

Returns

Type	Description
dict[str, Any]	MCP-format prompt response with ``description`` and ``messages``.

Raises

Exception	Description
MCPPromptError	If no prompt with the given name is registered.

---

ControllerResourceProvider

source

MCPResourceProviderProtocol backed by a list of MCPController instances.

Aggregates all @resource-decorated methods, exposes them for listing, and routes read_resource calls via URI pattern matching.

__init__

def __init__(controller_instances: list[MCPController]) -> None

source

Initialize with a list of already-constructed controller instances.

Parameters

Parameter	Type	Description
`controller_instances`	list[MCPController]	Instantiated MCPController objects to aggregate.

list_resources

async def list_resources() -> list[dict[str, Any]]

source

List all resources from all registered controllers in MCP format.

Returns

Type	Description
list[dict[str, Any]]	MCP resource definitions with ``uri``, ``name``, ``description``.

read_resource

async def read_resource(uri: str) -> dict[str, Any]

source

Read a resource by URI, matching exact URIs and {var} patterns.

Parameters

Parameter	Type	Description
`uri`	str	Concrete URI from the MCP client (e.g. ``"users://123"``).

Returns

Type	Description
dict[str, Any]	Resource content dict.

Raises

Exception	Description
MCPResourceError	If no resource pattern matches the given URI.

list_templates

async def list_templates() -> list[dict[str, Any]]

source

List URI templates for resources that contain {var} placeholders.

Returns

Type	Description
list[dict[str, Any]]	MCP URI template definitions with ``uriTemplate``, ``name``, ``description``.

---

ControllerToolProvider

source

MCPToolProviderProtocol backed by a list of MCPController instances.

Aggregates all @tool-decorated methods across multiple controllers, auto-generates JSON Schema from type annotations, and dispatches call_tool to the matching handler.

__init__

def __init__(controller_instances: list[MCPController]) -> None

source

Initialize with a list of already-constructed controller instances.

Parameters

Parameter	Type	Description
`controller_instances`	list[MCPController]	Instantiated MCPController objects to aggregate.

list_tools

async def list_tools() -> list[dict[str, Any]]

source

List all tools from all registered controllers in MCP format.

Returns

Type	Description
list[dict[str, Any]]	MCP tool definitions with ``name``, ``description``, ``inputSchema``.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any]
) -> Any

source

Dispatch a tool call to the matching controller method.

Parameters

Parameter	Type	Description
`name`	str	MCP tool name.
`arguments`	dict[str, Any]	Argument dict from the MCP client.

Returns

Type	Description
Any	Handler return value.

Raises

Exception	Description
MCPToolCallError	If no tool with the given name is registered.

---

MCPClient

source

Client for connecting to and calling tools on external MCP servers.

Manages the MCP protocol handshake (initialize / initialized notification) and exposes ergonomic methods for the most common MCP operations. The underlying MCPClientTransport handles the I/O; see StdioClientTransport and SSEClientTransport.

Supports async context manager usage for automatic lifecycle management

async with MCPClient(transport) as client:
    tools = await client.list_tools()

async with MCPClient(transport) as client:
    tools = await client.list_tools()

Parameters

Parameter	Type	Description
`transport`		The transport to use for server communication.
`request_timeout`		Per-request wait timeout in seconds. If a response is not received within this window a :exc:`~lexigram.ai.mcp.exceptions.MCPTransportError` is raised.

__init__

def __init__(
    transport: MCPClientTransport,
    *,
    request_timeout: float = 30.0
) -> None

source

connect

async def connect() -> None

source

Connect the transport and perform the MCP initialize handshake.

Raises

Exception	Description
MCPInitializationError	If the server rejects the handshake.
MCPTransportError	On transport-level failures.

disconnect

async def disconnect() -> None

source

Close the transport connection.

list_tools

async def list_tools() -> list[dict[str, Any]]

source

Return the list of tools exposed by the server.

Returns

Type	Description
list[dict[str, Any]]	List of MCP tool definition dicts, each with ``name``, ``description``, and ``inputSchema``.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any] | None = None
) -> Any

source

Call a tool by name on the server.

Parameters

Parameter	Type	Description
`name`	str	Tool name as returned by list_tools.
`arguments`	dict[str, Any] | None	Tool input arguments. Pass ``None`` for tools that take no parameters.

Returns

Type	Description
Any	The ``result`` field from the server's JSON-RPC response.

Raises

Exception	Description
MCPToolCallError	If the server returns an error for the tool call.

list_resources

async def list_resources() -> list[dict[str, Any]]

source

Return the list of resources exposed by the server.

Returns

Type	Description
list[dict[str, Any]]	List of MCP resource dicts, each with ``uri``, ``name``, ``description``, and ``mimeType``.

list_prompts

async def list_prompts() -> list[dict[str, Any]]

source

Return the list of prompt templates exposed by the server.

Returns

Type	Description
list[dict[str, Any]]	List of MCP prompt dicts, each with ``name``, ``description``, and ``arguments``.

---

MCPClientModule

source

Module for consuming external MCP servers via DI-injected clients.

Registers one or more MCPClient instances (wrapped in an MCPClientRegistry) so any service in the application can call external MCP tools.

Use configure to configure connections

app = LexigramApplication(
    modules=[
        MCPClientModule.configure(
            connections=[
                MCPConnection.stdio(
                    ["uvx", "mcp-server-git"],
                    name="git",
                ),
            ]
        ),
    ],
)

app = LexigramApplication(
    modules=[
        MCPClientModule.configure(
            connections=[
                MCPConnection.stdio(
                    ["uvx", "mcp-server-git"],
                    name="git",
                ),
            ]
        ),
    ],
)

__init__

def __init__(connections: list[MCPConnection] | None = None) -> None

source

configure

def configure(
    cls,
    connections: list[MCPConnection]
) -> MCPClientModule

source

Create an MCPClientModule for the given connections.

Parameters

Parameter	Type	Description
`connections`	list[MCPConnection]	List of MCPConnection objects describing each external MCP server. Each must have a unique *name*.

Returns

Type	Description
MCPClientModule	An ``MCPClientModule`` ready to be added to the application.

Raises

Exception	Description
ValueError	If *connections* is empty or contains duplicate names.

Example

MCPClientModule.configure(
    connections=[
        MCPConnection.stdio(["npx", "github-mcp"], name="github"),
        MCPConnection.sse("http://localhost:9000/mcp", name="local"),
    ],
)

MCPClientModule.configure(
    connections=[
        MCPConnection.stdio(["npx", "github-mcp"], name="github"),
        MCPConnection.sse("http://localhost:9000/mcp", name="local"),
    ],
)

providers

def providers() -> list

source

Return the DI providers for this module.

---

MCPClientProvider

source

__init__

def __init__(connections: list[MCPConnection]) -> None

source

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register MCPClientRegistry (and a single MCPClient if one connection).

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Boot hook for MCP client provider.

shutdown

async def shutdown() -> None

source

Shutdown MCP client provider resources.

---

MCPClientRegistry

source

Registry of named MCPClient instances.

Registered in the DI container by MCPClientModule. Inject this in services that need to call multiple external MCP servers

class ReportService:
    def __init__(self, registry: MCPClientRegistry) -> None:
        self._git = registry.get("git")
        self._analytics = registry.get("analytics")

class ReportService:
    def __init__(self, registry: MCPClientRegistry) -> None:
        self._git = registry.get("git")
        self._analytics = registry.get("analytics")

Call MCPClient.connect before making requests (or use the client as an async context manager for automatic lifecycle management).

__init__

def __init__(clients: dict[str, MCPClient]) -> None

source

Initialize with a pre-built mapping of names → clients.

Parameters

Parameter	Type	Description
`clients`	dict[str, MCPClient]	Dict mapping each connection name to its MCPClient instance.

get

def get(name: str) -> MCPClient

source

Return the client registered under name.

Parameters

Parameter	Type	Description
`name`	str	The connection name passed to MCPConnection.stdio or MCPConnection.sse.

Returns

Type	Description
MCPClient	The corresponding MCPClient.

Raises

Exception	Description
KeyError	If no client was registered with the given *name*.

names

def names() -> list[str]

source

Return the names of all registered connections.

Returns

Type	Description
list[str]	Sorted list of registered connection names.

---

MCPConfig

source

Configuration for the MCP server.

Attributes: host: Host to bind to (for HTTP transport). port: Port to bind to (for HTTP transport). path: URL path for MCP endpoint (default /mcp). enable_sse: Enable Server-Sent Events for streaming responses. stdio_mode: Use stdio transport instead of HTTP. server_name: Name of the MCP server. server_version: Version of the MCP server. cors_origins: CORS allowed origins (for HTTP transport).

---

MCPConnection

source

Immutable description of one external MCP server connection.

Do not construct directly — use the stdio or sse class methods.

Attributes: name: Unique identifier for this connection used by MCPClientRegistry. transport_type: "stdio" or "sse". command: (stdio only) Subprocess command to launch the MCP server. url: (sse only) Base HTTP URL of the MCP server. env: (stdio only) Optional extra environment variables. headers: (sse only) Optional extra HTTP headers. request_timeout: Per-request timeout in seconds (default 30). startup_timeout: (stdio only) Wait timeout for process start (default 10).

stdio

def stdio(
    cls,
    command: list[str],
    *,
    name: str,
    env: dict[str, str] | None = None,
    startup_timeout: float = 10.0,
    request_timeout: float = 30.0
) -> MCPConnection

source

Create a connection that communicates with a stdio-based MCP server.

Spawns command as a subprocess and exchanges newline-delimited JSON-RPC messages over its stdin/stdout.

Parameters

Parameter	Type	Description
`command`	list[str]	The command and arguments to launch the MCP server.
`name`	str	Unique name for use with MCPClientRegistry.
`env`	dict[str, str] | None	Optional extra environment variables.
`startup_timeout`	float	Seconds to wait for process start.
`request_timeout`	float	Per-request timeout in seconds.

Returns

Type	Description
MCPConnection	A configured ``MCPConnection`` describing this stdio server.

Example

MCPConnection.stdio(
    ["uvx", "mcp-server-git", "--repository", "/repo"],
    name="git",
)

MCPConnection.stdio(
    ["uvx", "mcp-server-git", "--repository", "/repo"],
    name="git",
)

sse

def sse(
    cls,
    url: str,
    *,
    name: str,
    headers: dict[str, str] | None = None,
    request_timeout: float = 30.0
) -> MCPConnection

source

Create a connection to an HTTP+SSE MCP server.

Parameters

Parameter	Type	Description
`url`	str	Base URL of the MCP server (e.g. ``http://localhost:8080/mcp``).
`name`	str	Unique name for use with MCPClientRegistry.
`headers`	dict[str, str] | None	Optional extra HTTP headers (e.g. auth tokens).
`request_timeout`	float	Per-request timeout in seconds.

Returns

Type	Description
MCPConnection	A configured ``MCPConnection`` describing this SSE server.

Example

MCPConnection.sse(
    "http://analytics.internal/mcp",
    name="analytics",
    headers={"Authorization": "Bearer secret"},
)

MCPConnection.sse(
    "http://analytics.internal/mcp",
    name="analytics",
    headers={"Authorization": "Bearer secret"},
)

build_transport

def build_transport() -> Any

source

Instantiate and return the appropriate transport for this connection.

Returns

Type	Description
Any	A StdioClientTransport or SSEClientTransport instance.

Raises

Exception	Description
ValueError	If *transport_type* is not ``"stdio"`` or ``"sse"``.

build_client

def build_client() -> MCPClient

source

Build an MCPClient from this connection description.

Returns

Type	Description
MCPClient	An unconnected MCPClient ready for use.

---

MCPController

source

Base class for MCP controllers.

Groups related MCP tools, resources, and prompt templates into a single class — mirroring the Controller DX from lexigram-web. Register decorated subclasses via MCPModule(controllers=[...]).

Example

class DataToolsController(MCPController):
    def __init__(self, repo: UserRepository) -> None:
        self.repo = repo

    @tool("get_user", description="Fetch a user by ID")
    async def get_user(self, user_id: str) -> dict:
        result = await self.repo.get(user_id)
        return result.unwrap_or({})

class DataToolsController(MCPController):
    def __init__(self, repo: UserRepository) -> None:
        self.repo = repo

    @tool("get_user", description="Fetch a user by ID")
    async def get_user(self, user_id: str) -> dict:
        result = await self.repo.get(user_id)
        return result.unwrap_or({})

collect_tools

def collect_tools(cls) -> list[dict[str, Any]]

source

Collect tool-decorated methods from the class hierarchy.

Returns

Type	Description
list[dict[str, Any]]	List of dicts with keys ``name``, ``description``, ``handler_name``.

collect_resources

def collect_resources(cls) -> list[dict[str, Any]]

source

Collect resource-decorated methods from the class hierarchy.

Returns

Type	Description
list[dict[str, Any]]	List of dicts with keys ``uri_pattern``, ``name``, ``description``, ``handler_name``.

collect_prompts

def collect_prompts(cls) -> list[dict[str, Any]]

source

Collect prompt-decorated methods from the class hierarchy.

Returns

Type	Description
list[dict[str, Any]]	List of dicts with keys ``name``, ``description``, ``handler_name``.

---

MCPInitializeResult

source

Result of the initialize method.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPJSONRPCRequest

source

JSON-RPC request message.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to JSON-RPC format.

---

MCPJSONRPCResponse

source

JSON-RPC response message.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to JSON-RPC format.

create_success

def create_success(
    result: dict[str, Any],
    request_id: int | str | None = None
) -> MCPJSONRPCResponse

source

Create a success response.

create_error

def create_error(
    code: int,
    message: str,
    request_id: int | str | None = None,
    data: Any | None = None
) -> MCPJSONRPCResponse

source

Create an error response.

---

MCPModule

source

Module for MCP server.

Provides MCP server, handlers, and transports to the application.

Usage

app = LexigramApplication(
    modules=[..., MCPModule.configure()],
)

# With MCPController subclasses::

app = LexigramApplication(
    modules=[..., MCPModule.configure(controllers=[DataToolsController])],
)

# Auto-expose existing services as MCP tools::

app = LexigramApplication(
    modules=[
        ...,
        MCPModule.from_services(
            services=[UserService, AnalyticsService],
            include_methods=["search", "get_*"],
        ),
    ],
)

app = LexigramApplication(
    modules=[..., MCPModule.configure()],
)

# With MCPController subclasses

app = LexigramApplication(
    modules=[..., MCPModule.configure(controllers=[DataToolsController])],
)

# Auto-expose existing services as MCP tools::

app = LexigramApplication(
    modules=[
        ...,
        MCPModule.from_services(
            services=[UserService, AnalyticsService],
            include_methods=["search", "get_*"],
        ),
    ],
)

app = LexigramApplication(
    modules=[..., MCPModule.configure(controllers=[DataToolsController])],
)

# Auto-expose existing services as MCP tools

app = LexigramApplication(
    modules=[
        ...,
        MCPModule.from_services(
            services=[UserService, AnalyticsService],
            include_methods=["search", "get_*"],
        ),
    ],
)

app = LexigramApplication(
    modules=[
        ...,
        MCPModule.from_services(
            services=[UserService, AnalyticsService],
            include_methods=["search", "get_*"],
        ),
    ],
)

from_services

def from_services(
    cls,
    services: list[type],
    *,
    include_methods: list[str] | None = None,
    config: MCPConfig | None = None
) -> DynamicModule

source

Create an MCPModule that auto-exposes service methods as MCP tools.

Parameters

Parameter	Type	Description
`services`	list[type]	Service classes whose public methods are exposed as tools.
`include_methods`	list[str] | None	Optional glob patterns to restrict which methods are exposed (e.g. ``["search", "get_*"]``).
`config`	MCPConfig | None	Optional MCPConfig override.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

configure

def configure(
    cls,
    *,
    config: MCPConfig | None = None,
    controllers: list[type] | None = None,
    services: list[type] | None = None,
    include_methods: list[str] | None = None,
    enable_streaming: bool = True,
    **provider_kwargs: Any
) -> DynamicModule

source

Create an MCPModule with explicit provider configuration.

Parameters

Parameter	Type	Description
`config`	MCPConfig | None	Optional MCPConfig.
`controllers`	list[type] | None	MCPController subclasses to register with the server.
`services`	list[type] | None	Service classes to auto-expose as MCP tools.
`include_methods`	list[str] | None	Glob patterns to restrict auto-exposed methods.
`enable_streaming`	bool	Enable SSE streaming transport support. Defaults to ``True``; set to ``False`` to restrict the server to stdio-only transport. **provider_kwargs: Additional keyword arguments forwarded to MCPProvider.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: MCPConfig | None = None
) -> DynamicModule

source

Create an MCPModule suitable for unit and integration testing.

Uses in-memory transport with no external MCP server connections. Streaming is disabled by default to simplify test assertions.

Parameters

Parameter	Type	Description
`config`	MCPConfig | None	Optional MCPConfig override. Uses safe test defaults when ``None``.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

MCPPrompt

source

An MCP prompt template that clients can use.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPPromptMessage

source

A message in an MCP prompt response.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPProvider

source

__init__

def __init__(
    config: MCPConfig | None = None,
    controllers: list[type] | None = None,
    services: list[type] | None = None,
    include_methods: list[str] | None = None,
    enable_streaming: bool = True,
    **provider_kwargs: Any
) -> None

source

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register MCPConfig with the DI container.

boot

async def boot(container: BootContainerProtocol) -> None

source

Wire up all MCP components using the resolved container.

shutdown

async def shutdown() -> None

source

Clean up MCP provider resources.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Return a health check result for this provider.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for health response.

Returns

Type	Description
HealthCheckResult	A HealthCheckResult reflecting the current MCP provider state.

---

MCPResource

source

An MCP resource that clients can browse and read.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPResourceContent

source

Content of an MCP resource returned to the client.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPServer

source

Core MCP server — routes JSON-RPC messages to handlers.

The server is transport-agnostic. It receives parsed JSON-RPC messages and returns JSON-RPC responses. The transport layer (stdio or HTTP+SSE) handles serialization and I/O.

Usage

server = MCPServer(
    name="my-app",
    tool_handler=tool_handler,
    resource_handler=resource_handler,
)
response = await server.handle_message(request)

server = MCPServer(
    name="my-app",
    tool_handler=tool_handler,
    resource_handler=resource_handler,
)
response = await server.handle_message(request)

__init__

def __init__(
    config: Any | None = None,
    name: str | None = None,
    version: str | None = None,
    tool_handler: Any | None = None,
    resource_handler: Any | None = None,
    prompt_handler: Any | None = None,
    sampling_handler: Any | None = None,
    logging_handler: Any | None = None
) -> None

source

Initialize the MCP server.

Parameters

Parameter	Type	Description
`config`	Any | None	Optional MCP server configuration.
`name`	str | None	Server name (overrides config if provided).
`version`	str | None	Server version (overrides config if provided).
`tool_handler`	Any | None	Handler for tool-related methods.
`resource_handler`	Any | None	Handler for resource-related methods.
`prompt_handler`	Any | None	Handler for prompt-related methods.
`sampling_handler`	Any | None	Handler for sampling/createMessage (optional).
`logging_handler`	Any | None	Handler for logging/setLevel (optional).

handle_message

async def handle_message(message: dict[str, Any]) -> dict[str, Any] | None

source

Handle a JSON-RPC message and return the response.

Parameters

Parameter	Type	Description
`message`	dict[str, Any]	Parsed JSON-RPC request message.

Returns

Type	Description
dict[str, Any] | None	JSON-RPC response dict, or None for notifications.

---

MCPServerCapabilities

source

Server capabilities sent during initialization.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPServerInfo

source

Server metadata sent during initialization.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPServerStartedHook

source

Payload fired after the MCP server completes its startup sequence.

Attributes: transport: Transport type in use (e.g. "stdio" or "sse").

---

MCPServerStoppedHook

source

Payload fired after the MCP server shuts down.

Attributes: transport: Transport type that was in use at shutdown.

---

MCPToolDefinition

source

MCP tool definition sent to clients during tools/list.

Maps directly to the MCP tool schema format.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to MCP protocol format.

---

MCPToolInvokedHook

source

Payload fired each time a tool is invoked through the MCP protocol.

Attributes: tool_name: Name of the MCP tool that was called.

---

ModulePromptProvider

source

Prompt provider built from module-level ``@prompt``-decorated functions.

Use from_module to scan a Python module for decorated callables.

__init__

def __init__(funcs: list[Any]) -> None

source

Initialize with a list of decorated functions.

Parameters

Parameter	Type	Description
`funcs`	list[Any]	Callables that have ``_prompt_config`` set by ``@prompt()``.

from_module

def from_module(
    cls,
    module: Any
) -> ModulePromptProvider

source

Scan module for @prompt-decorated functions.

Parameters

Parameter	Type	Description
`module`	Any	A Python module object.

Returns

Type	Description
ModulePromptProvider	A new ``ModulePromptProvider`` with all discovered prompts.

list_prompts

async def list_prompts() -> list[dict[str, Any]]

source

Return all registered prompt definitions.

get_prompt

async def get_prompt(
    name: str,
    arguments: dict[str, Any] | None = None
) -> Any

source

Return a prompt by name.

Parameters

Parameter	Type	Description
`name`	str	Prompt name.
`arguments`	dict[str, Any] | None	Optional template arguments.

Returns

Type	Description
Any	MCP prompt response dict.

Raises

Exception	Description
MCPPromptError	If the prompt name is not registered.

---

ModuleResourceProvider

source

Resource provider built from module-level ``@resource``-decorated functions.

Use from_module to scan a Python module for decorated callables.

__init__

def __init__(funcs: list[Any]) -> None

source

Initialize with a list of decorated functions.

Parameters

Parameter	Type	Description
`funcs`	list[Any]	Callables that have ``_resource_config`` set by ``@resource()``.

from_module

def from_module(
    cls,
    module: Any
) -> ModuleResourceProvider

source

Scan module for @resource-decorated functions.

Parameters

Parameter	Type	Description
`module`	Any	A Python module object.

Returns

Type	Description
ModuleResourceProvider	A new ``ModuleResourceProvider`` with all discovered resources.

list_resources

async def list_resources() -> list[dict[str, Any]]

source

Return all registered resource definitions.

read_resource

async def read_resource(uri: str) -> Any

source

Read a resource by URI.

Parameters

Parameter	Type	Description
`uri`	str	Concrete URI from the MCP client.

Returns

Type	Description
Any	Resource content.

Raises

Exception	Description
MCPResourceError	If no resource pattern matches the given URI.

list_templates

async def list_templates() -> list[dict[str, Any]]

source

Return URI templates (resources with {var} placeholders).

---

ModuleToolProvider

source

Tool provider built from module-level ``@tool``-decorated functions.

Use from_module to scan a Python module for decorated callables

import my_tools
provider = ModuleToolProvider.from_module(my_tools)

import my_tools
provider = ModuleToolProvider.from_module(my_tools)

Or pass functions directly

provider = ModuleToolProvider([search_users, create_user])

provider = ModuleToolProvider([search_users, create_user])

__init__

def __init__(funcs: list[Any]) -> None

source

Initialize with a list of decorated functions.

Parameters

Parameter	Type	Description
`funcs`	list[Any]	Callables that have ``_tool_config`` set by ``@tool()``.

from_module

def from_module(
    cls,
    module: Any
) -> ModuleToolProvider

source

Scan module for @tool-decorated functions.

Parameters

Parameter	Type	Description
`module`	Any	A Python module object.

Returns

Type	Description
ModuleToolProvider	A new ``ModuleToolProvider`` with all discovered tools.

list_tools

async def list_tools() -> list[dict[str, Any]]

source

Return MCP tool definitions for all registered functions.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any]
) -> Any

source

Invoke the tool function by name.

Parameters

Parameter	Type	Description
`name`	str	Tool name.
`arguments`	dict[str, Any]	Arguments dict from the MCP client.

Returns

Type	Description
Any	Tool result.

Raises

Exception	Description
MCPToolCallError	If the tool name is not registered.

---

PromptHandler

source

Handler for MCP prompt-related methods.

Handles prompts/list and prompts/get methods by delegating to an MCPPromptProviderProtocol implementation.

__init__

def __init__(prompt_provider: Any | None = None) -> None

source

Initialize the prompt handler.

Parameters

Parameter	Type	Description
`prompt_provider`	Any | None	Provider that handles prompt operations.

list_prompts

async def list_prompts() -> Result[dict[str, Any], MCPError]

source

Handle prompts/list method.

Returns

Type	Description
Result[dict[str, Any], MCPError]	``Result`` containing prompts list in MCP format.

get_prompt

async def get_prompt(
    name: str,
    arguments: dict[str, Any] | None = None
) -> Result[dict[str, Any], MCPError]

source

Handle prompts/get method.

Parameters

Parameter	Type	Description
`name`	str	Name of the prompt to get.
`arguments`	dict[str, Any] | None	Arguments to fill in the prompt template.

Returns

Type	Description
Result[dict[str, Any], MCPError]	``Result[dict[str, Any], MCPError]`` with prompt payload.

---

ResourceHandler

source

Handler for MCP resource-related methods.

Handles resources/list, resources/read, and resources/templates/list methods by delegating to an MCPResourceProviderProtocol implementation.

__init__

def __init__(resource_provider: Any | None = None) -> None

source

Initialize the resource handler.

Parameters

Parameter	Type	Description
`resource_provider`	Any | None	Provider that handles resource operations.

list_resources

async def list_resources() -> Result[dict[str, Any], MCPError]

source

Handle resources/list method.

Returns

Type	Description
Result[dict[str, Any], MCPError]	``Result`` containing resources list in MCP format.

read_resource

async def read_resource(uri: str) -> Result[dict[str, Any], MCPError]

source

Handle resources/read method.

Parameters

Parameter	Type	Description
`uri`	str	URI of the resource to read.

Returns

Type	Description
Result[dict[str, Any], MCPError]	``Result[dict[str, Any], MCPError]`` with MCP-formatted contents.

list_templates

async def list_templates() -> Result[dict[str, Any], MCPError]

source

Handle resources/templates/list method.

Returns

Type	Description
Result[dict[str, Any], MCPError]	``Result`` containing URI templates in MCP format.

---

SSEClientTransport

source

MCP client transport that communicates with an HTTP+SSE MCP server.

Sends JSON-RPC requests as HTTP POST bodies and reads the responses from the response body. Uses aiohttp for async HTTP — ensure it is installed (pip install aiohttp or the http extra of this package).

Parameters

Parameter	Type	Description
`url`		Base URL of the MCP HTTP endpoint (e.g. ``http://localhost:8080/mcp``).
`headers`		Additional HTTP headers to include with every request (e.g. ``{"Authorization": "Bearer "}``.
`request_timeout`		Per-request timeout in seconds.

__init__

def __init__(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    request_timeout: float = 30.0
) -> None

source

connect

async def connect() -> None

source

Open an aiohttp session.

disconnect

async def disconnect() -> None

source

Close the aiohttp session.

send

async def send(message: dict[str, Any]) -> None

source

Queue message for the next receive call via HTTP POST.

receive

async def receive() -> dict[str, Any]

source

POST the pending request and return the parsed JSON-RPC response.

---

SSETransport

source

Server-Sent Events transport for MCP server.

This transport uses HTTP POST for requests and SSE for streaming responses. Ideal for web-based integrations.

__init__

def __init__(server: Any | None = None) -> None

source

Initialize the SSE transport.

Parameters

Parameter	Type	Description
`server`	Any | None	Optional HTTP server instance.

start

async def start() -> None

source

Start the SSE transport.

stop

async def stop() -> None

source

Stop the SSE transport.

send

async def send(message: dict[str, Any]) -> None

source

Send a message via SSE.

Parameters

Parameter	Type	Description
`message`	dict[str, Any]	JSON-RPC message to send.

Raises

Exception	Description
MCPTransportError	If sending fails.

receive

async def receive() -> dict[str, Any] | None

source

Receive is not applicable for SSE (pull-based).

For HTTP+SSE, use the HTTP endpoint directly instead.

Returns

Type	Description
dict[str, Any] | None	None (receiving is handled via HTTP POST).

get_queued_messages

def get_queued_messages() -> list[dict[str, Any]]

source

Get queued messages for SSE delivery.

Returns

Type	Description
list[dict[str, Any]]	List of queued JSON-RPC messages.

---

ServiceToolProvider

source

Tool provider that auto-exposes public methods of service instances as MCP tools.

Created via MCPModule.from_services or directly

provider = ServiceToolProvider.from_services(
    [user_service, analytics_service],
    include_patterns=["search", "get_*"],
)

provider = ServiceToolProvider.from_services(
    [user_service, analytics_service],
    include_patterns=["search", "get_*"],
)

Tool names are prefixed with the lowercase service class name to avoid collisions (e.g. UserService.search_users() → "userservice_search_users"). The first line of each method’s docstring is used as the tool description. Both async def and plain def methods are supported.

__init__

def __init__(entries: list[tuple[Any, str, str, str]]) -> None

source

Initialize with pre-built tool entries.

Parameters

Parameter	Type	Description
`entries`	list[tuple[Any, str, str, str]]	List of ``(instance, method_name, tool_name, description)`` tuples.

from_services

def from_services(
    cls,
    service_instances: Any,
    include_patterns: list[str] | None = None
) -> ServiceToolProvider

source

Scan service_instances and expose matching public methods as tools.

Parameters

Parameter	Type	Description
`service_instances`	Any	A single instantiated service object or a list.
`include_patterns`	list[str] | None	Glob patterns to filter method names (e.g. ``["search", "get_*"]``). When ``None``, all public non-dunder methods are included.

Returns

Type	Description
ServiceToolProvider	A new ``ServiceToolProvider`` with entries for every matched method.

list_tools

async def list_tools() -> list[dict[str, Any]]

source

Return MCP tool definitions for all discovered service methods.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any]
) -> Any

source

Invoke the service method mapped to name.

Parameters

Parameter	Type	Description
`name`	str	Tool name.
`arguments`	dict[str, Any]	Arguments dict from the MCP client.

Returns

Type	Description
Any	Method return value (sync or awaited coroutine).

Raises

Exception	Description
MCPToolCallError	If no method is registered under *name*.

---

StdioClientTransport

source

MCP client transport that communicates with a subprocess via stdio.

Spawns command as an asyncio subprocess and exchanges newline-delimited JSON-RPC messages over its stdin/stdout pipes.

Parameters

Parameter	Type	Description
`command`		The command and arguments to launch the MCP server process.
`env`		Optional extra environment variables to pass to the subprocess. If *None*, the current process environment is inherited.
`startup_timeout`		Seconds to wait for the process to start before
`raising`		exc:`~lexigram.ai.mcp.exceptions.MCPTransportError`.

__init__

def __init__(
    command: list[str],
    *,
    env: dict[str, str] | None = None,
    startup_timeout: float = 10.0
) -> None

source

connect

async def connect() -> None

source

Spawn the subprocess.

disconnect

async def disconnect() -> None

source

Terminate the subprocess.

send

async def send(message: dict[str, Any]) -> None

source

Write a newline-delimited JSON message to the subprocess stdin.

receive

async def receive() -> dict[str, Any]

source

Read a newline-delimited JSON message from subprocess stdout.

---

StdioTransport

source

Stdio-based transport for MCP server.

This transport reads JSON-RPC messages from stdin and writes responses to stdout. Ideal for CLI/desktop integration.

__init__

def __init__(
    reader: asyncio.StreamReader | None = None,
    writer: asyncio.StreamWriter | None = None
) -> None

source

Initialize the stdio transport.

Parameters

Parameter	Type	Description
`reader`	asyncio.StreamReader | None	Optional stream reader (for testing).
`writer`	asyncio.StreamWriter | None	Optional stream writer (for testing).

start

async def start() -> None

source

Start the stdio transport.

stop

async def stop() -> None

source

Stop the stdio transport.

send

async def send(message: dict[str, Any]) -> None

source

Send a message through stdio.

Parameters

Parameter	Type	Description
`message`	dict[str, Any]	JSON-RPC message to send.

Raises

Exception	Description
MCPTransportError	If sending fails.

receive

async def receive() -> dict[str, Any] | None

source

Receive a message from stdio.

Returns

Type	Description
dict[str, Any] | None	Parsed JSON-RPC message, or None if no data available.

Raises

Exception	Description
MCPTransportError	If receiving fails.

---

ToolHandler

source

Handler for MCP tool-related methods.

Handles tools/list and tools/call methods by delegating to a MCPToolProviderProtocol implementation.

All handler methods return Result so callers can handle expected failures without raising exceptions.

__init__

def __init__(tool_provider: Any | None = None) -> None

source

Initialize the tool handler.

Parameters

Parameter	Type	Description
`tool_provider`	Any | None	Provider that handles tool listing and execution.

list_tools

async def list_tools() -> Result[dict[str, Any], MCPError]

source

Handle tools/list method.

Returns

Type	Description
Result[dict[str, Any], MCPError]	``Result`` containing tools list in MCP format.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any] | None = None
) -> Result[dict[str, Any], MCPError]

source

Handle tools/call method.

Returns a Result — Ok wraps the serialized MCP tool result dict; Err wraps an MCPToolCallError describing what went wrong.

Parameters

Parameter	Type	Description
`name`	str	Name of the tool to call.
`arguments`	dict[str, Any] | None	Arguments to pass to the tool.

Returns

Type	Description
Result[dict[str, Any], MCPError]	``Result[dict[str, Any], MCPError]`` — never raises expected errors.

---

ToolRegistryAdapter

source

Adapter that bridges lexigram-agents ToolRegistry to MCPToolProviderProtocol.

This adapter allows the MCP server to use the @tool-decorated functions from lexigram-agents. The translation is minimal:

• ToolProtocol.parameters_schema → MCP inputSchema
• ToolRegistryProtocol.get() → lookup by name
• ToolProtocol.execute(**kw) → tool invocation

__init__

def __init__(tool_registry: ToolRegistryProtocol | None = None) -> None

source

list_tools

async def list_tools() -> list[dict[str, Any]]

source

List all available tools in MCP format.

Returns

Type	Description
list[dict[str, Any]]	List of tool definitions with ``name``, ``description``, and ``inputSchema`` (JSON Schema produced by ``@tool``).

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any]
) -> Any

source

Execute a tool by name with given arguments.

Parameters

Parameter	Type	Description
`name`	str	Name of the tool to call.
`arguments`	dict[str, Any]	Arguments to pass to the tool.

Returns

Type	Description
Any	Tool execution result.

---

Functions

prompt

prompt

def prompt(
    name: str,
    *,
    description: str = ''
) -> Any

source

Register a method as an MCP prompt template handler.

Stores _prompt_config on the decorated method for discovery by collect_prompts.

Parameters

Parameter	Type	Description
`name`	str	MCP prompt name exposed to AI clients.
`description`	str	Human-readable description of the prompt template.

Returns

Type	Description
Any	Decorator that attaches prompt metadata to the method.

Example

@prompt("summarize", description="Generate a summary prompt")
async def summarize_prompt(self) -> str:
    return "Summarize the following data..."

@prompt("summarize", description="Generate a summary prompt")
async def summarize_prompt(self) -> str:
    return "Summarize the following data..."

---

resource

resource

def resource(
    uri_pattern: str,
    *,
    description: str = '',
    name: str | None = None
) -> Any

source

Register a method as an MCP resource handler.

Stores _resource_config on the decorated method for discovery by collect_resources.

Parameters

Parameter	Type	Description
`uri_pattern`	str	URI template (e.g. ``"users://{user_id}"``).
`description`	str	Human-readable description of the resource.
`name`	str | None	Optional display name; defaults to the URI pattern.

Returns

Type	Description
Any	Decorator that attaches resource metadata to the method.

Example

@resource("users://{user_id}", description="Expose user by ID")
async def get_user(self, user_id: str) -> dict:
    ...

@resource("users://{user_id}", description="Expose user by ID")
async def get_user(self, user_id: str) -> dict:
    ...

---

tool

tool

def tool(
    name: str,
    *,
    description: str = ''
) -> Any

source

Register a method as an MCP tool handler.

Stores _tool_config on the decorated method for discovery by collect_tools.

Parameters

Parameter	Type	Description
`name`	str	MCP tool name exposed to AI clients.
`description`	str	Human-readable description for the tool schema.

Returns

Type	Description
Any	Decorator that attaches tool metadata to the method.

Example

@tool("search_users", description="Search users by name")
async def search_users(self, name: str = "") -> list[dict]:
    ...

@tool("search_users", description="Search users by name")
async def search_users(self, name: str = "") -> list[dict]:
    ...

---

Exceptions

MCPError

source

Base exception for all MCP errors.

__init__

def __init__(
    message: str = 'MCP error',
    **kwargs: Any
) -> None

source

---

MCPInitializationError

source

MCP server initialization failed.

Raised when the MCP server cannot be initialized, usually due to missing required handlers or configuration errors.

__init__

def __init__(
    message: str = 'MCP initialization failed',
    *,
    reason: str | None = None,
    **kwargs: Any
) -> None

source

---

MCPMethodNotFoundError

source

Unknown MCP method requested by client.

Raised when the client requests an MCP method that the server does not support.

__init__

def __init__(
    message: str = 'MCP method not found',
    *,
    method: str | None = None,
    **kwargs: Any
) -> None

source

---

MCPPromptError

source

Prompt retrieval or list failed.

Raised when a prompt operation fails, such as when a prompt is not found or arguments are invalid.

__init__

def __init__(
    message: str = 'MCP prompt error',
    *,
    prompt_name: str | None = None,
    **kwargs: Any
) -> None

source

---

MCPProtocolError

source

Protocol violation (malformed message, invalid state).

Raised when an MCP message violates the protocol specification, such as missing required fields or invalid JSON-RPC structure.

__init__

def __init__(
    message: str = 'MCP protocol error',
    *,
    details: dict[str, Any] | None = None,
    **kwargs: Any
) -> None

source

---

MCPResourceError

source

Resource read or list failed.

Raised when a resource operation fails, such as when a resource is not found or cannot be read.

__init__

def __init__(
    message: str = 'MCP resource error',
    *,
    uri: str | None = None,
    **kwargs: Any
) -> None

source

---

MCPToolCallError

source

Tool call failed during MCP execution.

Raised when a tool invocation fails, either due to the tool not existing or raising an exception.

__init__

def __init__(
    message: str = 'MCP tool call failed',
    *,
    tool_name: str | None = None,
    **kwargs: Any
) -> None

source

---