windlass.providers.mcp.fastmcp

fastmcp

FastMCP client adapter, plus a dependency-free in-process client for tests.

Model Context Protocol servers expose tools, resources and prompts over stdio or HTTP. Once connected, their tools become ordinary Windlass tools — an agent binding a local Python function and a remote MCP tool cannot tell them apart, which is exactly the point.

Install with::

pip install "windlass[mcp]"

Example

from windlass import Windlass # doctest: +SKIP agent = Windlass.agent().mcp("filesystem", # doctest: +SKIP ... command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "."])

FastMCPClient

FastMCPClient(
    *,
    server: str = "",
    command: str | None = None,
    args: list[str] | None = None,
    url: str | None = None,
    env: dict[str, str] | None = None,
    config: dict[str, Any] | None = None,
    namespace: bool = False,
    timeout: float = 30.0,
    **kwargs: Any
)

Bases: MCPClient

MCP client built on fastmcp.

Parameters:

Name	Type	Description	Default
server	str	Label for this server, used in logs and namespacing.	''
command	str | None	Executable for a stdio server, e.g. "npx" or "python".	None
args	list[str] | None	Arguments for command.	None
url	str | None	HTTP or SSE endpoint, as an alternative to command.	None
env	dict[str, str] | None	Extra environment variables for a stdio server.	None
config	dict[str, Any] | None	A full FastMCP client configuration dict, for advanced setups.	None
namespace	bool	Prefix remote tool names with the server label, so two servers offering search do not collide.	False
timeout	float	Per-call timeout in seconds.	30.0
**kwargs	Any	Forwarded to :class:~windlass.interfaces.mcp.MCPClient.	{}

Raises:

Type	Description
MissingDependencyError	When fastmcp is not installed.
ConfigurationError	When neither command, url nor config is given.
MCPError	When the server cannot be reached.

Note

A stdio server is a subprocess this client starts and owns. Use it as an async context manager, or call :meth:~windlass.interfaces.mcp.MCPClient.disconnect, so the process is reaped.

Source code in src\windlass\providers\mcp\fastmcp.py

def __init__(
    self,
    *,
    server: str = "",
    command: str | None = None,
    args: list[str] | None = None,
    url: str | None = None,
    env: dict[str, str] | None = None,
    config: dict[str, Any] | None = None,
    namespace: bool = False,
    timeout: float = 30.0,
    **kwargs: Any,
) -> None:
    super().__init__(server=server or "mcp", namespace=namespace, timeout=timeout, **kwargs)
    self._fastmcp = require("fastmcp", extra="mcp", feature="The MCP client")
    if not (command or url or config):
        raise ConfigurationError(
            "An MCP client needs a transport.",
            hint="Pass command=... (+args) for a stdio server, url=... for HTTP, "
            "or config=... for a full FastMCP configuration.",
        )
    self.command = command
    self.args = list(args or [])
    self.url = url
    self.env = dict(env or {})
    self.server_config = config
    self._client: Any = None
    self._session: Any = None

native

native() -> Any

Return the underlying FastMCP client (Level 3 access).

Source code in src\windlass\providers\mcp\fastmcp.py

def native(self) -> Any:
    """Return the underlying FastMCP client (Level 3 access)."""
    return self._client

aconnect async

aconnect() -> None

Open the transport and enter the MCP session.

Idempotent: calling it twice does not start a second server.

Raises:

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

Source code in src\windlass\providers\mcp\fastmcp.py

async def aconnect(self) -> None:
    """Open the transport and enter the MCP session.

    Idempotent: calling it twice does not start a second server.

    Raises:
        MCPError: When the server cannot be started or the handshake fails.
    """
    if self.connected:
        return
    try:
        self._client = self._fastmcp.Client(self._transport())
        self._session = await self._client.__aenter__()
        self.connected = True
    except Exception as exc:
        self._client = self._session = None
        raise MCPError(
            f"Could not connect to MCP server {self.server!r}: {exc}",
            hint="Check the command is on PATH (try running it by hand), or that "
            "the URL is reachable.",
            context={"server": self.server, "command": self.command, "url": self.url},
        ) from exc

adisconnect async

adisconnect() -> None

Close the session and reap any subprocess.

Source code in src\windlass\providers\mcp\fastmcp.py

async def adisconnect(self) -> None:
    """Close the session and reap any subprocess."""
    if self._client is not None:
        try:
            await self._client.__aexit__(None, None, None)
        except Exception as exc:
            self._log.debug("MCP disconnect from %s failed: %s", self.server, exc)
    self._client = self._session = None
    self.connected = False

alist_tools async

alist_tools() -> list[Tool]

Discover the server's tools.

Returns:

Type	Description
list[Tool]	class:~windlass.interfaces.mcp.MCPToolProxy instances, ready to
list[Tool]	bind to an agent.

Raises:

Type	Description
MCPError	When discovery fails.

Source code in src\windlass\providers\mcp\fastmcp.py

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

    Returns:
        :class:`~windlass.interfaces.mcp.MCPToolProxy` instances, ready to
        bind to an agent.

    Raises:
        MCPError: When discovery fails.
    """
    await self.aconnect()
    try:
        discovered = await self._client.list_tools()
    except Exception as exc:
        raise MCPError(
            f"Could not list tools on MCP server {self.server!r}: {exc}",
            context={"server": self.server},
        ) from exc

    tools: list[Tool] = []
    for item in discovered:
        remote_name = _attr(item, "name", "")
        if not remote_name:
            continue
        tools.append(
            MCPToolProxy(
                self,
                name=self._tool_name(remote_name),
                description=_attr(item, "description", "") or "",
                parameters=_attr(item, "inputSchema", None)
                or _attr(item, "input_schema", None)
                or {"type": "object", "properties": {}},
                server=self.server,
                timeout=self.timeout,
            )
        )
    return tools

acall_tool async

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

Invoke a remote tool.

Parameters:

Name	Type	Description	Default
name	str	Tool name, with any Windlass namespace prefix stripped.	required
arguments	dict[str, Any]	Arguments matching the remote schema.	required

Returns:

Type	Description
Any	The unwrapped result — plain text where the server returned text,
Any	parsed JSON where it returned JSON, otherwise the raw payload.

Raises:

Type	Description
MCPError	When the call fails.

Source code in src\windlass\providers\mcp\fastmcp.py

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

    Args:
        name: Tool name, with any Windlass namespace prefix stripped.
        arguments: Arguments matching the remote schema.

    Returns:
        The unwrapped result — plain text where the server returned text,
        parsed JSON where it returned JSON, otherwise the raw payload.

    Raises:
        MCPError: When the call fails.
    """
    await self.aconnect()
    remote = _strip_namespace(name, self.server) if self.namespace else name
    try:
        result = await self._client.call_tool(remote, arguments)
    except Exception as exc:
        raise MCPError(
            f"MCP tool {remote!r} on {self.server!r} failed: {exc}",
            context={"server": self.server, "tool": remote},
        ) from exc
    return _unwrap_content(result)

alist_resources async

alist_resources() -> list[MCPResource]

Discover the server's readable resources.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_resources(self) -> list[MCPResource]:
    """Discover the server's readable resources."""
    await self.aconnect()
    try:
        discovered = await self._client.list_resources()
    except Exception as exc:
        self._log.debug("MCP server %s does not support resources: %s", self.server, exc)
        return []
    return [
        MCPResource(
            uri=str(_attr(item, "uri", "")),
            name=_attr(item, "name", "") or "",
            description=_attr(item, "description", "") or "",
            mimetype=_attr(item, "mimeType", None) or _attr(item, "mimetype", None),
            server=self.server,
        )
        for item in discovered
        if _attr(item, "uri", None)
    ]

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\providers\mcp\fastmcp.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.
    """
    await self.aconnect()
    try:
        result = await self._client.read_resource(uri)
    except Exception as exc:
        raise MCPError(
            f"Could not read MCP resource {uri!r}: {exc}",
            context={"server": self.server, "uri": uri},
        ) from exc
    content = _unwrap_content(result)
    return content if isinstance(content, str) else json.dumps(content, default=str)

alist_prompts async

alist_prompts() -> list[MCPPrompt]

Discover the server's prompt templates.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_prompts(self) -> list[MCPPrompt]:
    """Discover the server's prompt templates."""
    await self.aconnect()
    try:
        discovered = await self._client.list_prompts()
    except Exception as exc:
        self._log.debug("MCP server %s does not support prompts: %s", self.server, exc)
        return []
    return [
        MCPPrompt(
            name=_attr(item, "name", ""),
            description=_attr(item, "description", "") or "",
            arguments=[_as_dict(a) for a in (_attr(item, "arguments", None) or [])],
            server=self.server,
        )
        for item in discovered
        if _attr(item, "name", None)
    ]

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\providers\mcp\fastmcp.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.
    """
    await self.aconnect()
    try:
        result = await self._client.get_prompt(name, arguments or {})
    except Exception as exc:
        raise MCPError(
            f"Could not render MCP prompt {name!r}: {exc}",
            context={"server": self.server, "prompt": name},
        ) from exc

    messages = _attr(result, "messages", None)
    if messages:
        parts = []
        for message in messages:
            content = _attr(message, "content", "")
            parts.append(content if isinstance(content, str) else _unwrap_content(content))
        return "\n\n".join(str(p) for p in parts if p)
    content = _unwrap_content(result)
    return content if isinstance(content, str) else json.dumps(content, default=str)

StaticMCPClient

StaticMCPClient(
    *,
    tools: dict[str, Callable[..., Any]] | None = None,
    resources: dict[str, str] | None = None,
    prompts: dict[str, str] | None = None,
    server: str = "static",
    namespace: bool = False,
    **kwargs: Any
)

Bases: MCPClient

An MCP client backed by local Python callables.

Lets you exercise every MCP code path — discovery, namespacing, tool proxying, agent binding — without starting a subprocess or a server. That makes MCP integration testable in ordinary unit tests.

Parameters:

Name	Type	Description	Default
tools	dict[str, Callable[..., Any]] | None	Mapping of tool name to callable.	None
resources	dict[str, str] | None	Mapping of URI to content.	None
prompts	dict[str, str] | None	Mapping of prompt name to a template string, formatted with the supplied arguments.	None
server	str	Server label.	'static'
namespace	bool	Prefix tool names with the server label.	False
**kwargs	Any	Forwarded to :class:~windlass.interfaces.mcp.MCPClient.	{}

Example

client = StaticMCPClient( ... tools={"shout": lambda text: text.upper()}, ... resources={"file://greeting": "hello"}, ... ) client.connect() client.call_tool("shout", {"text": "hi"}) 'HI' client.read_resource("file://greeting") 'hello'

Source code in src\windlass\providers\mcp\fastmcp.py

def __init__(
    self,
    *,
    tools: dict[str, Callable[..., Any]] | None = None,
    resources: dict[str, str] | None = None,
    prompts: dict[str, str] | None = None,
    server: str = "static",
    namespace: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(server=server, namespace=namespace, **kwargs)
    self._tools = dict(tools or {})
    self._resources = dict(resources or {})
    self._prompts = dict(prompts or {})

aconnect async

aconnect() -> None

Mark the client connected. There is no transport to open.

Source code in src\windlass\providers\mcp\fastmcp.py

async def aconnect(self) -> None:
    """Mark the client connected. There is no transport to open."""
    self.connected = True

alist_tools async

alist_tools() -> list[Tool]

Return one tool per registered callable, with derived schemas.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_tools(self) -> list[Tool]:
    """Return one tool per registered callable, with derived schemas."""
    await self.aconnect()
    built: list[Tool] = []
    for name, fn in self._tools.items():
        wrapped = FunctionTool(fn, name=self._tool_name(name))
        built.append(
            MCPToolProxy(
                self,
                name=wrapped.name,
                description=wrapped.description,
                parameters=wrapped.parameters,
                server=self.server,
            )
        )
    return built

acall_tool async

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

Invoke a registered callable.

Parameters:

Name	Type	Description	Default
name	str	Tool name, with any namespace prefix stripped.	required
arguments	dict[str, Any]	Keyword arguments.	required

Returns:

Type	Description
Any	The callable's return value.

Raises:

Type	Description
MCPError	When no such tool is registered or the call fails.

Source code in src\windlass\providers\mcp\fastmcp.py

async def acall_tool(self, name: str, arguments: dict[str, Any]) -> Any:
    """Invoke a registered callable.

    Args:
        name: Tool name, with any namespace prefix stripped.
        arguments: Keyword arguments.

    Returns:
        The callable's return value.

    Raises:
        MCPError: When no such tool is registered or the call fails.
    """
    remote = _strip_namespace(name, self.server) if self.namespace else name
    fn = self._tools.get(remote)
    if fn is None:
        raise MCPError(
            f"No tool named {remote!r} on server {self.server!r}.",
            hint=f"Available tools: {', '.join(sorted(self._tools)) or '(none)'}",
            context={"available": sorted(self._tools)},
        )
    import asyncio

    try:
        if asyncio.iscoroutinefunction(fn):
            return await fn(**arguments)
        return fn(**arguments)
    except Exception as exc:
        raise MCPError(f"MCP tool {remote!r} failed: {exc}") from exc

alist_resources async

alist_resources() -> list[MCPResource]

Return the registered resources.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_resources(self) -> list[MCPResource]:
    """Return the registered resources."""
    return [
        MCPResource(uri=uri, name=uri.rsplit("/", 1)[-1], server=self.server)
        for uri in self._resources
    ]

aread_resource async

aread_resource(uri: str) -> str

Return a registered resource's content.

Raises:

Type	Description
MCPError	When the URI is not registered.

Source code in src\windlass\providers\mcp\fastmcp.py

async def aread_resource(self, uri: str) -> str:
    """Return a registered resource's content.

    Raises:
        MCPError: When the URI is not registered.
    """
    if uri not in self._resources:
        raise MCPError(
            f"No resource at {uri!r}.",
            hint=f"Available resources: {', '.join(sorted(self._resources)) or '(none)'}",
            context={"available": sorted(self._resources)},
        )
    return self._resources[uri]

alist_prompts async

alist_prompts() -> list[MCPPrompt]

Return the registered prompt templates.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_prompts(self) -> list[MCPPrompt]:
    """Return the registered prompt templates."""
    return [MCPPrompt(name=name, server=self.server) for name in self._prompts]

aget_prompt async

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

Render a registered prompt template.

Raises:

Type	Description
MCPError	When the prompt is not registered or a placeholder is missing.

Source code in src\windlass\providers\mcp\fastmcp.py

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

    Raises:
        MCPError: When the prompt is not registered or a placeholder is missing.
    """
    template = self._prompts.get(name)
    if template is None:
        raise MCPError(
            f"No prompt named {name!r}.", context={"available": sorted(self._prompts)}
        )
    try:
        return template.format(**(arguments or {}))
    except KeyError as exc:
        raise MCPError(f"Prompt {name!r} is missing argument {exc}.") from exc

MultiMCPClient

MultiMCPClient(
    *, clients: list[MCPClient] | None = None, server: str = "multi", **kwargs: Any
)

Bases: MCPClient

Aggregates several MCP servers behind one client.

Tool discovery unions every server's tools, and calls are routed back to whichever server advertised the tool. Namespacing is forced on, because two servers offering search is the normal case, not the exception.

Parameters:

Name	Type	Description	Default
clients	list[MCPClient] | None	The servers to aggregate.	None
server	str	Label for the aggregate.	'multi'
**kwargs	Any	Forwarded to :class:~windlass.interfaces.mcp.MCPClient.	{}

Example

a = StaticMCPClient(tools={"ping": lambda: "pong"}, server="alpha") b = StaticMCPClient(tools={"ping": lambda: "pong"}, server="beta") multi = MultiMCPClient(clients=[a, b]) sorted(t.name for t in multi.list_tools()) ['alpha_ping', 'beta_ping']

Source code in src\windlass\providers\mcp\fastmcp.py

def __init__(
    self, *, clients: list[MCPClient] | None = None, server: str = "multi", **kwargs: Any
) -> None:
    super().__init__(server=server, namespace=True, **kwargs)
    self.clients = list(clients or [])
    for client in self.clients:
        client.namespace = True
    self._routes: dict[str, MCPClient] = {}

add

add(client: MCPClient) -> MultiMCPClient

Add a server and return self.

Source code in src\windlass\providers\mcp\fastmcp.py

def add(self, client: MCPClient) -> MultiMCPClient:
    """Add a server and return ``self``."""
    client.namespace = True
    self.clients.append(client)
    return self

native

native() -> Any

Return the list of underlying clients.

Source code in src\windlass\providers\mcp\fastmcp.py

def native(self) -> Any:
    """Return the list of underlying clients."""
    return self.clients

aconnect async

aconnect() -> None

Connect every server, tolerating individual failures.

Source code in src\windlass\providers\mcp\fastmcp.py

async def aconnect(self) -> None:
    """Connect every server, tolerating individual failures."""
    from windlass.core.concurrency import gather_bounded

    if not self.clients:
        self.connected = True
        return
    outcomes = await gather_bounded(
        [c.aconnect() for c in self.clients],
        limit=len(self.clients),
        return_exceptions=True,
    )
    for client, outcome in zip(self.clients, outcomes, strict=True):
        if isinstance(outcome, BaseException):
            self._log.warning("MCP server %s unavailable: %s", client.server, outcome)
    self.connected = True

adisconnect async

adisconnect() -> None

Disconnect every server.

Source code in src\windlass\providers\mcp\fastmcp.py

async def adisconnect(self) -> None:
    """Disconnect every server."""
    for client in self.clients:
        await client.adisconnect()
    self.connected = False

alist_tools async

alist_tools() -> list[Tool]

Union every server's tools, recording where each one came from.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_tools(self) -> list[Tool]:
    """Union every server's tools, recording where each one came from."""
    await self.aconnect()
    tools: list[Tool] = []
    for client in self.clients:
        try:
            discovered = await client.alist_tools()
        except Exception as exc:
            self._log.warning("Tool discovery failed on %s: %s", client.server, exc)
            continue
        for item in discovered:
            self._routes[item.name] = client
            tools.append(item)
    return tools

acall_tool async

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

Route a call to the server that advertised the tool.

Raises:

Type	Description
MCPError	When no server owns the tool.

Source code in src\windlass\providers\mcp\fastmcp.py

async def acall_tool(self, name: str, arguments: dict[str, Any]) -> Any:
    """Route a call to the server that advertised the tool.

    Raises:
        MCPError: When no server owns the tool.
    """
    client = self._routes.get(name)
    if client is None:
        await self.alist_tools()
        client = self._routes.get(name)
    if client is None:
        raise MCPError(
            f"No MCP server provides a tool named {name!r}.",
            context={"available": sorted(self._routes)},
        )
    return await client.acall_tool(name, arguments)

alist_resources async

alist_resources() -> list[MCPResource]

Union every server's resources.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_resources(self) -> list[MCPResource]:
    """Union every server's resources."""
    await self.aconnect()
    resources: list[MCPResource] = []
    for client in self.clients:
        try:
            resources.extend(await client.alist_resources())
        except Exception as exc:
            self._log.debug("Resource listing failed on %s: %s", client.server, exc)
    return resources

alist_prompts async

alist_prompts() -> list[MCPPrompt]

Union every server's prompts.

Source code in src\windlass\providers\mcp\fastmcp.py

async def alist_prompts(self) -> list[MCPPrompt]:
    """Union every server's prompts."""
    await self.aconnect()
    prompts: list[MCPPrompt] = []
    for client in self.clients:
        try:
            prompts.extend(await client.alist_prompts())
        except Exception as exc:
            self._log.debug("Prompt listing failed on %s: %s", client.server, exc)
    return prompts