duck.contrib.mcp.capabilities.defaults

Default capabilities for the MCPView.

Module Contents

Classes

ElicitationCapability	MCP elicitation capability.
InitializeCapability	The initialize capability for the MCP view.
NotificationsCapability	Utility capability for sending MCP notifications.
PingCapability	The ping capability for the MCP view.
PromptsCapability	The prompts capability for the MCP view.
ResourcesCapability	The resources capability for the MCP view.
RootsCapability	MCP roots capability.
SamplingCapability	MCP sampling capability.
ServerRequestsCapability	Capability that enables server-to-client request/response via hooks.
ToolsCapability	The tools capability for the MCP view.

API

class duck.contrib.mcp.capabilities.defaults.ElicitationCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

MCP elicitation capability.

Provides helpers for requesting user input from the client through the MCP elicitation/create request.

This capability does not expose any handlers because elicitation is a server-to-client request initiated by the server.

Initialization

async create(message: str, requested_schema: dict, *, timeout: float = 60.0) → dict

Request information from the user through the client.

Parameters:

• message – Message displayed to the user.

• requested_schema – JSON schema describing the information being requested.

• timeout – Maximum time to wait for the client response.

Returns:

Client elicitation response.

Raises:

• TimeoutError – If the client does not respond within the timeout.

• MCPError – If the client returns an error.

property server_requests: duck.contrib.mcp.capabilities.Capability

Access the internal server request capability.

setup()

Setup the capability.

class duck.contrib.mcp.capabilities.defaults.InitializeCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

The initialize capability for the MCP view.

Initialization

async initialize(params)

Handle the initialize handshake: report server identity, protocol version, and capabilities.

setup()

Setup the capability.

class duck.contrib.mcp.capabilities.defaults.NotificationsCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

Utility capability for sending MCP notifications.

This is not advertised in the MCP initialize capabilities object because notifications are part of the protocol layer, not a negotiated capability.

Initialization

async handle_initialized(params: dict)

Handle the MCP notifications/initialized notification.

This notification is sent by the client after it has successfully completed the initialize handshake. It does not return a response.

async notify(method: str, params: Optional[dict] = None) → None

Send a notification to the connected client.

Parameters:

• method – Notification method name.

• params – Optional notification parameters.

setup()

Setup the capability.

suppress_response()

Drop the current request without sending a response.

Used for JSON-RPC notifications and other requests where the protocol requires no response body.

class duck.contrib.mcp.capabilities.defaults.PingCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

The ping capability for the MCP view.

Initialization

async ping(params)

Handle the MCP ping request.

MCP uses ping as a lightweight liveness check. No state changes are required and the response body is an empty result object.

setup()

Setup the capability.

class duck.contrib.mcp.capabilities.defaults.PromptsCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

The prompts capability for the MCP view.

Initialization

property prompts: Dict[str, Callable]

Get all prompts from decorated MCPView methods.

async prompts_get(params)

Handle prompts/get: render the named prompt with the given arguments into MCP message format.

async prompts_list(params)

Handle prompts/list: return name and description for every registered prompt.

setup()

Setup the capability.

class duck.contrib.mcp.capabilities.defaults.ResourcesCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

The resources capability for the MCP view.

Initialization

match_resource_template(uri: str)

Match a URI against registered resource templates.

Returns:

(handler, extracted_arguments)

returns:

(None, {}) if no template matches.

Return type:

tuple

property resources: Dict[str, Callable]

Get all resources from decorated MCPView methods.

async resources_list(params)

Handle resources/list: return uri, description, and mime type for every registered resource.

async resources_read(params)

Handle resources/read: fetch static or templated resources.

setup()

Setup the capability.

property templates: Dict[str, Callable]

Get all resource templates from decorated MCPView methods.

async templates_list(params)

Handle resources/templates/list.

class duck.contrib.mcp.capabilities.defaults.RootsCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

MCP roots capability.

Provides helpers for requesting the client to list available filesystem roots through the MCP roots/list request.

This capability does not expose any MCP handlers because roots are fetched by the server from the client.

Initialization

async list_roots(timeout: float = 10.0) → list

Request available roots from the client.

Parameters:

timeout – Maximum time to wait for the client response.

Returns:

A list of root objects.

Raises:

• TimeoutError – If the client does not respond within the timeout.

• MCPError – If the client returns an error.

property server_requests: duck.contrib.mcp.capabilities.Capability

Access the internal server request capability.

setup()

Setup the capability.

class duck.contrib.mcp.capabilities.defaults.SamplingCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

MCP sampling capability.

Provides helpers for requesting the client to generate model completions through the MCP sampling/createMessage request.

This capability does not expose any MCP handlers because sampling is a server-to-client request initiated by the server.

Initialization

async create_message(messages: list, max_tokens: int, *, model_preferences: Optional[dict] = None, system_prompt: Optional[str] = None, include_context: Optional[str] = None, temperature: Optional[float] = None, stop_sequences: Optional[list[str]] = None, timeout: float = 60.0) → dict

Request the client to sample a model response.

Parameters:

• messages – Conversation messages in MCP format.

• max_tokens – Maximum number of tokens the client may generate.

• model_preferences – Optional model selection hints.

• system_prompt – Optional system instruction.

• include_context – Optional context inclusion preference.

• temperature – Optional sampling temperature.

• stop_sequences – Optional generation stop sequences.

• timeout – Maximum time to wait for the client response.

Returns:

The client’s sampling response.

Raises:

• TimeoutError – If the client does not respond in time.

• MCPError – If the client returns an error.

property server_requests: duck.contrib.mcp.capabilities.Capability

Access the internal server request capability.

setup()

Setup the capability.

class duck.contrib.mcp.capabilities.defaults.ServerRequestsCapability(*args, **kwargs)

Bases: duck.contrib.mcp.capabilities.Capability

Capability that enables server-to-client request/response via hooks.

This capability registers itself with the view’s hook system to intercept incoming client responses. It provides the send_request() method that tools can use to send JSON-RPC requests to the client and await responses.

The capability automatically:

• Registers a pre-request hook to detect client responses

• Manages pending request futures with timeouts

• Cleans up on capability removal

Example:

class MyServer(MCPView):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.register_capability("_server_requests", ServerRequestsCapability(self), alias="server")

    @tool(description="Ask client for input")
    async def ask_client(self, question: str) -> str:
        result = await self._server_requests.send_request(
            "custom/ask",
            {"question": question}
        )
        return result.get("answer", "No answer")

Initialization

cleanup()

Cancel all pending requests and release any associated resources.

async handle_client_response(body: dict, rpc_id, method, params)

Hook that intercepts incoming messages to detect client responses.

If the message is a response (has ‘id’ and either ‘result’ or ‘error’, but no ‘method’), it resolves the pending future and returns a special sentinel to short-circuit further processing.

Returns:

tuple (handled, result_or_response)

• (True, response): Message was a client response, return this to client

• (False, None): Not a response, continue normal processing

property pending_requests: Dict[int, asyncio.Future]

Return the pending requests in queue.

async send_request(method: str, params: Optional[dict] = None, timeout: float = 10.0) → Any

Send a JSON-RPC request to the client and wait for its response.

Parameters:

• method – The method name to call on the client.

• params – Optional parameters for the method.

• timeout – Maximum seconds to wait for a response.

Returns:

The result field from the client’s response.

Raises:

• RuntimeError – If no active SSE stream exists.

• TimeoutError – If the client does not respond within the timeout.

• Exception – If the client returns an error object.

setup()

Register hooks and initialize state.

class duck.contrib.mcp.capabilities.defaults.ToolsCapability(view: MCPView)

Bases: duck.contrib.mcp.capabilities.Capability

The tools capability for the MCP view.

Initialization

setup()

Setup the capability.

property tools: Dict[str, Callable]

Get all tools from decorated MCPView methods.

async tools_call(params)

Handle tools/call: invoke the named tool with the given arguments and wrap its result as MCP content.

async tools_list(params)

Handle tools/list: return name, description, and input schema for every registered tool.