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 quadkit-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:~quadkit.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 = QuadkitApplication(
    modules=[
        MCPClientModule.configure(
            connections=[
                MCPConnection.stdio(
                    ["uvx", "mcp-server-git"],
                    name="git",
                ),
            ]
        ),
    ],
)

app = QuadkitApplication(
    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://localhost:8080/mcp",
    name="analytics",
    headers={"Authorization": "Bearer secret"},
)

MCPConnection.sse(
    "http://localhost:8080/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 quadkit-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 = QuadkitApplication(
    modules=[..., MCPModule.configure()],
)

# With MCPController subclasses::

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

# Auto-expose existing services as MCP tools::

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

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

# With MCPController subclasses

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

# Auto-expose existing services as MCP tools::

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

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

# Auto-expose existing services as MCP tools

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

app = QuadkitApplication(
    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,
    authorizer: MCPAuthorizerProtocol | None = None,
    allow_unauthenticated: bool = False
) -> 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).
authorizer	MCPAuthorizerProtocol | None	Optional authz hook consulted once per non-handshake request post-initialize. `None means requests are denied with -32000 unless allow_unauthenticated` is set.
allow_unauthenticated	bool	When `True (and no authorizer) restore the open posture: non-handshake methods dispatch without authorization. Defaults to False` (fail-closed).

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:~quadkit.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 quadkit-agents ToolRegistry to MCPToolProviderProtocol.

This adapter allows the MCP server to use the @tool-decorated functions from quadkit-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

---