windlass.interfaces.mcp

mcp

The Model Context Protocol (MCP) interface.

MCP servers expose three things: tools an agent can call, resources it can read, and prompts it can instantiate. Windlass treats an MCP server as one more source of tools — after connecting, its tools appear alongside your local Python functions and the agent cannot tell them apart.

Implementers override :meth:MCPClient.aconnect, :meth:MCPClient.alist_tools and :meth:MCPClient.acall_tool; resources and prompts are optional.

Example

from windlass.providers.mcp.fastmcp import StaticMCPClient client = StaticMCPClient(tools={"echo": lambda text: text}) client.connect() [t.name for t in client.list_tools()]['echo']

MCPResource

Bases: WindlassModel

A readable resource advertised by an MCP server.

Attributes:

Name	Type	Description
uri	str	Resource identifier, e.g. file:///docs/readme.md.
name	str	Human readable name.
description	str	What the resource contains.
mimetype	str | None	Content type, when the server declares one.
server	str	Which server advertised it.

MCPPrompt

Bases: WindlassModel

A parameterised prompt template advertised by an MCP server.

Attributes:

Name	Type	Description
name	str	Prompt identifier.
description	str	What the prompt is for.
arguments	list[dict[str, Any]]	Declared arguments, each a dict with name, description and required.
server	str	Which server advertised it.

MCPToolProxy

MCPToolProxy(
    client: MCPClient,
    *,
    name: str,
    description: str = "",
    parameters: dict[str, Any] | None = None,
    server: str = "",
    **config: Any
)

Bases: Tool

A local :class:~windlass.interfaces.tool.Tool backed by a remote MCP tool.

Created by :meth:MCPClient.alist_tools. Calling it forwards to the server and returns whatever comes back, so agents bind it exactly like a local function.

Parameters:

Name	Type	Description	Default
client	MCPClient	The MCP client that owns the connection.	required
name	str	Remote tool name.	required
description	str	Remote tool description.	''
parameters	dict[str, Any] | None	Remote JSON Schema for the arguments.	None
server	str	Server label, kept in metadata and used to disambiguate same-named tools across servers.	''
**config	Any	Forwarded to :class:~windlass.interfaces.tool.Tool.	{}

Source code in src\windlass\interfaces\mcp.py

def __init__(
    self,
    client: MCPClient,
    *,
    name: str,
    description: str = "",
    parameters: dict[str, Any] | None = None,
    server: str = "",
    **config: Any,
) -> None:
    super().__init__(
        name=name,
        description=description or f"MCP tool {name!r} from {server or 'server'}.",
        parameters=parameters,
        **config,
    )
    self.client = client
    self.server = server

acall async

acall(**kwargs: Any) -> Any

Forward the call to the MCP server.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arguments matching the remote schema.	{}

Returns:

Type	Description
Any	The server's result.

Raises:

Type	Description
MCPError	When the call fails or the server is unreachable.

Source code in src\windlass\interfaces\mcp.py

async def acall(self, **kwargs: Any) -> Any:
    """Forward the call to the MCP server.

    Args:
        **kwargs: Arguments matching the remote schema.

    Returns:
        The server's result.

    Raises:
        MCPError: When the call fails or the server is unreachable.
    """
    return await self.client.acall_tool(self.name, kwargs)

native

native() -> Any

Return the underlying MCP client session.

Source code in src\windlass\interfaces\mcp.py

def native(self) -> Any:
    """Return the underlying MCP client session."""
    return self.client.native()

MCPClient

MCPClient(
    *,
    server: str = "",
    namespace: bool = False,
    timeout: float = 30.0,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract MCP client.

Parameters:

Name	Type	Description	Default
server	str	Server label used in logs and to namespace tools.	''
namespace	bool	When True, remote tools are exposed as server_toolname so two servers offering search do not collide.	False
timeout	float	Per-call timeout in seconds.	30.0
name	str | None	Component name.	None
**config	Any	Transport-specific options (command, args, url, env, ...).	{}

Attributes:

Name	Type	Description
connected		Whether the transport is currently open.

Example

Implementing a client means three methods::

class MyClient(MCPClient):
    provider_name = "mine"

    async def aconnect(self): ...
    async def alist_tools(self): ...
    async def acall_tool(self, name, arguments): ...

Source code in src\windlass\interfaces\mcp.py

def __init__(
    self,
    *,
    server: str = "",
    namespace: bool = False,
    timeout: float = 30.0,
    name: str | None = None,
    **config: Any,
) -> None:
    super().__init__(
        name=name or server or self.provider_name,
        server=server,
        namespace=namespace,
        timeout=timeout,
        **config,
    )
    self.server = server or self.provider_name
    self.namespace = namespace
    self.timeout = timeout
    self.connected = False

aconnect abstractmethod async

aconnect() -> None

Open the transport and perform the MCP handshake.

Must be idempotent — calling it twice should not open two sessions.

Raises:

Type	Description
MCPError	When the server cannot be reached or the handshake fails.

Source code in src\windlass\interfaces\mcp.py

@abc.abstractmethod
async def aconnect(self) -> None:
    """Open the transport and perform the MCP handshake.

    Must be idempotent — calling it twice should not open two sessions.

    Raises:
        MCPError: When the server cannot be reached or the handshake fails.
    """

alist_tools abstractmethod async

alist_tools() -> list[Tool]

Discover the server's tools.

Returns:

Type	Description
list[Tool]	class:MCPToolProxy instances ready to bind to an agent.

Raises:

Type	Description
MCPError	When discovery fails.

Source code in src\windlass\interfaces\mcp.py

@abc.abstractmethod
async def alist_tools(self) -> list[Tool]:
    """Discover the server's tools.

    Returns:
        :class:`MCPToolProxy` instances ready to bind to an agent.

    Raises:
        MCPError: When discovery fails.
    """

acall_tool abstractmethod async

acall_tool(name: str, arguments: dict[str, Any]) -> Any

Invoke a remote tool.

Parameters:

Name	Type	Description	Default
name	str	Tool name as advertised by the server (without any namespace prefix Windlass may have added).	required
arguments	dict[str, Any]	Arguments matching the remote schema.	required

Returns:

Type	Description
Any	The server's result, unwrapped from the MCP content envelope.

Raises:

Type	Description
MCPError	When the call fails.

Source code in src\windlass\interfaces\mcp.py

@abc.abstractmethod
async def acall_tool(self, name: str, arguments: dict[str, Any]) -> Any:
    """Invoke a remote tool.

    Args:
        name: Tool name as advertised by the server (without any namespace
            prefix Windlass may have added).
        arguments: Arguments matching the remote schema.

    Returns:
        The server's result, unwrapped from the MCP content envelope.

    Raises:
        MCPError: When the call fails.
    """

alist_resources async

alist_resources() -> list[MCPResource]

Discover readable resources. Returns [] when unsupported.

Source code in src\windlass\interfaces\mcp.py

async def alist_resources(self) -> list[MCPResource]:
    """Discover readable resources. Returns ``[]`` when unsupported."""
    return []

aread_resource async

aread_resource(uri: str) -> str

Read a resource's contents.

Parameters:

Name	Type	Description	Default
uri	str	The resource identifier.	required

Returns:

Type	Description
str	The resource body as text.

Raises:

Type	Description
MCPError	When the resource cannot be read.

Source code in src\windlass\interfaces\mcp.py

async def aread_resource(self, uri: str) -> str:
    """Read a resource's contents.

    Args:
        uri: The resource identifier.

    Returns:
        The resource body as text.

    Raises:
        MCPError: When the resource cannot be read.
    """
    raise MCPError(
        f"{type(self).__name__} does not support reading resources.",
        context={"uri": uri},
    )

alist_prompts async

alist_prompts() -> list[MCPPrompt]

Discover prompt templates. Returns [] when unsupported.

Source code in src\windlass\interfaces\mcp.py

async def alist_prompts(self) -> list[MCPPrompt]:
    """Discover prompt templates. Returns ``[]`` when unsupported."""
    return []

aget_prompt async

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

Instantiate a prompt template.

Parameters:

Name	Type	Description	Default
name	str	Prompt name.	required
arguments	dict[str, Any] | None	Template arguments.	None

Returns:

Type	Description
str	The rendered prompt text.

Raises:

Type	Description
MCPError	When the prompt cannot be rendered.

Source code in src\windlass\interfaces\mcp.py

async def aget_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> str:
    """Instantiate a prompt template.

    Args:
        name: Prompt name.
        arguments: Template arguments.

    Returns:
        The rendered prompt text.

    Raises:
        MCPError: When the prompt cannot be rendered.
    """
    raise MCPError(f"{type(self).__name__} does not support prompts.", context={"prompt": name})

adisconnect async

adisconnect() -> None

Close the transport. Safe to call when already closed.

Source code in src\windlass\interfaces\mcp.py

async def adisconnect(self) -> None:
    """Close the transport. Safe to call when already closed."""
    self.connected = False

connect

connect() -> None

Blocking :meth:aconnect.

Source code in src\windlass\interfaces\mcp.py

def connect(self) -> None:
    """Blocking :meth:`aconnect`."""
    run_sync(self.aconnect())

list_tools

list_tools() -> list[Tool]

Blocking :meth:alist_tools.

Source code in src\windlass\interfaces\mcp.py

def list_tools(self) -> list[Tool]:
    """Blocking :meth:`alist_tools`."""
    return run_sync(self.alist_tools())

call_tool

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

Blocking :meth:acall_tool.

Source code in src\windlass\interfaces\mcp.py

def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> Any:
    """Blocking :meth:`acall_tool`."""
    return run_sync(self.acall_tool(name, arguments or {}))

list_resources

list_resources() -> list[MCPResource]

Blocking :meth:alist_resources.

Source code in src\windlass\interfaces\mcp.py

def list_resources(self) -> list[MCPResource]:
    """Blocking :meth:`alist_resources`."""
    return run_sync(self.alist_resources())

read_resource

read_resource(uri: str) -> str

Blocking :meth:aread_resource.

Source code in src\windlass\interfaces\mcp.py

def read_resource(self, uri: str) -> str:
    """Blocking :meth:`aread_resource`."""
    return run_sync(self.aread_resource(uri))

list_prompts

list_prompts() -> list[MCPPrompt]

Blocking :meth:alist_prompts.

Source code in src\windlass\interfaces\mcp.py

def list_prompts(self) -> list[MCPPrompt]:
    """Blocking :meth:`alist_prompts`."""
    return run_sync(self.alist_prompts())

get_prompt

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

Blocking :meth:aget_prompt.

Source code in src\windlass\interfaces\mcp.py

def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> str:
    """Blocking :meth:`aget_prompt`."""
    return run_sync(self.aget_prompt(name, arguments))

disconnect

disconnect() -> None

Blocking :meth:adisconnect.

Source code in src\windlass\interfaces\mcp.py

def disconnect(self) -> None:
    """Blocking :meth:`adisconnect`."""
    run_sync(self.adisconnect())

aclose async

aclose() -> None

Alias for :meth:adisconnect, so containers can close uniformly.

Source code in src\windlass\interfaces\mcp.py

async def aclose(self) -> None:
    """Alias for :meth:`adisconnect`, so containers can close uniformly."""
    await self.adisconnect()