fastmcp_client

langroid/agent/tools/mcp/fastmcp_client.py

FastMCPClient(server, sampling_handler=None, roots=None, log_handler=None, message_handler=None, read_timeout_seconds=None)

A client for interacting with a FastMCP server.

Provides async context manager functionality to safely manage resources.

Parameters:

Name	Type	Description	Default
server	FastMCPServerSpec	FastMCP server or path to such a server	required

Source code in langroid/agent/tools/mcp/fastmcp_client.py

def __init__(
    self,
    server: FastMCPServerSpec,
    sampling_handler: SamplingHandler | None = None,  # type: ignore
    roots: RootsList | RootsHandler | None = None,  # type: ignore
    log_handler: LoggingFnT | None = None,
    message_handler: MessageHandlerFnT | None = None,
    read_timeout_seconds: datetime.timedelta | None = None,
) -> None:
    """Initialize the FastMCPClient.

    Args:
        server: FastMCP server or path to such a server
    """
    self.server = server
    self.client = None
    self._cm = None
    self.sampling_handler = sampling_handler
    self.roots = roots
    self.log_handler = log_handler
    self.message_handler = message_handler
    self.read_timeout_seconds = read_timeout_seconds

connect() async

Open the underlying session.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def connect(self) -> None:
    """Open the underlying session."""
    await self.__aenter__()

close() async

Close the underlying session.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def close(self) -> None:
    """Close the underlying session."""
    await self.__aexit__(None, None, None)

get_tool_async(tool_name) async

Create a Langroid ToolMessage subclass from the MCP Tool with the given tool_name.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def get_tool_async(self, tool_name: str) -> Type[ToolMessage]:
    """
    Create a Langroid ToolMessage subclass from the MCP Tool
    with the given `tool_name`.
    """
    if not self.client:
        raise RuntimeError("Client not initialized. Use async with FastMCPClient.")
    target = await self.get_mcp_tool_async(tool_name)
    if target is None:
        raise ValueError(f"No tool named {tool_name}")
    props = target.inputSchema.get("properties", {})
    fields: Dict[str, Tuple[type, Any]] = {}
    for fname, schema in props.items():
        ftype, fld = self._schema_to_field(fname, schema, target.name)
        fields[fname] = (ftype, fld)

    # Convert target.name to CamelCase and add Tool suffix
    parts = target.name.replace("-", "_").split("_")
    camel_case = "".join(part.capitalize() for part in parts)
    model_name = f"{camel_case}Tool"

    from langroid.agent.tool_message import ToolMessage as _BaseToolMessage

    # IMPORTANT: Avoid clashes with reserved field names in Langroid ToolMessage!
    # First figure out which field names are reserved
    reserved = set(_BaseToolMessage.__annotations__.keys())
    reserved.update(["recipient", "_handler", "name"])
    renamed: Dict[str, str] = {}
    new_fields: Dict[str, Tuple[type, Any]] = {}
    for fname, (ftype, fld) in fields.items():
        if fname in reserved:
            new_name = fname + "__"
            renamed[fname] = new_name
            new_fields[new_name] = (ftype, fld)
        else:
            new_fields[fname] = (ftype, fld)
    # now replace fields with our renamed‐aware mapping
    fields = new_fields

    # create Langroid ToolMessage subclass, with expected fields.
    tool_model = cast(
        Type[ToolMessage],
        create_model(  # type: ignore[call-overload]
            model_name,
            request=(str, target.name),
            purpose=(str, target.description or f"Use the tool {target.name}"),
            __base__=ToolMessage,
            **fields,
        ),
    )
    # Store ALL client configuration needed to recreate a client
    client_config = {
        "server": self.server,
        "sampling_handler": self.sampling_handler,
        "roots": self.roots,
        "log_handler": self.log_handler,
        "message_handler": self.message_handler,
        "read_timeout_seconds": self.read_timeout_seconds,
    }

    tool_model._client_config = client_config  # type: ignore [attr-defined]
    tool_model._renamed_fields = renamed  # type: ignore[attr-defined]

    # 2) define an arg-free call_tool_async()
    async def call_tool_async(self: ToolMessage) -> Any:
        from langroid.agent.tools.mcp.fastmcp_client import FastMCPClient

        # pack up the payload
        payload = self.dict(
            exclude=self.Config.schema_extra["exclude"].union(
                ["request", "purpose"]
            ),
        )

        # restore any renamed fields
        for orig, new in self.__class__._renamed_fields.items():  # type: ignore
            if new in payload:
                payload[orig] = payload.pop(new)

        client_cfg = getattr(self.__class__, "_client_config", None)  # type: ignore
        if not client_cfg:
            # Fallback or error - ideally _client_config should always exist
            raise RuntimeError(f"Client config missing on {self.__class__}")
        # open a fresh client, call the tool, then close
        async with FastMCPClient(**client_cfg) as client:  # type: ignore
            return await client.call_mcp_tool(self.request, payload)

    tool_model.call_tool_async = call_tool_async  # type: ignore

    if not hasattr(tool_model, "handle_async"):
        # 3) define an arg-free handle_async() method
        # if the tool model doesn't already have one
        async def handle_async(self: ToolMessage) -> Any:
            return await self.call_tool_async()  # type: ignore[attr-defined]

        # add the handle_async() method to the tool model
        tool_model.handle_async = handle_async  # type: ignore

    return tool_model

get_tools_async() async

Get all available tools as Langroid ToolMessage classes, handling nested schemas, with handle_async methods

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def get_tools_async(self) -> List[Type[ToolMessage]]:
    """
    Get all available tools as Langroid ToolMessage classes,
    handling nested schemas, with `handle_async` methods
    """
    if not self.client:
        raise RuntimeError("Client not initialized. Use async with FastMCPClient.")
    resp = await self.client.list_tools()
    return [await self.get_tool_async(t.name) for t in resp]

get_mcp_tool_async(name) async

Find the "original" MCP Tool (i.e. of type mcp.types.Tool) on the server matching name, or None if missing. This contains the metadata for the tool: name, description, inputSchema, etc.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool to look up.	required

Returns:

Type	Description
Optional[Tool]	The raw Tool object from the server, or None.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def get_mcp_tool_async(self, name: str) -> Optional[Tool]:
    """Find the "original" MCP Tool (i.e. of type mcp.types.Tool) on the server
     matching `name`, or None if missing. This contains the metadata for the tool:
     name, description, inputSchema, etc.

    Args:
        name: Name of the tool to look up.

    Returns:
        The raw Tool object from the server, or None.
    """
    if not self.client:
        raise RuntimeError("Client not initialized. Use async with FastMCPClient.")
    resp: List[Tool] = await self.client.list_tools()
    return next((t for t in resp if t.name == name), None)

call_mcp_tool(tool_name, arguments) async

Call an MCP tool with the given arguments.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool to call.	required
arguments	Dict[str, Any]	Arguments to pass to the tool.	required

Returns:

Type	Description
str | List[str] | None	The result of the tool call.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def call_mcp_tool(
    self, tool_name: str, arguments: Dict[str, Any]
) -> str | List[str] | None:
    """Call an MCP tool with the given arguments.

    Args:
        tool_name: Name of the tool to call.
        arguments: Arguments to pass to the tool.

    Returns:
        The result of the tool call.
    """
    if not self.client:
        raise RuntimeError("Client not initialized. Use async with FastMCPClient.")
    result: CallToolResult = await self.client.session.call_tool(
        tool_name,
        arguments,
    )
    return self._convert_tool_result(tool_name, result)

get_tool_async(server, tool_name, **client_kwargs) async

Get a single Langroid ToolMessage subclass for a specific MCP tool name (async).

This is a convenience wrapper that creates a temporary FastMCPClient.

Parameters:

Name	Type	Description	Default
server	FastMCPServerSpec	Specification of the FastMCP server to connect to.	required
tool_name	str	The name of the tool to retrieve.	required
**client_kwargs	Any	Additional keyword arguments to pass to the FastMCPClient constructor (e.g., sampling_handler, roots).	{}

Returns:

Type	Description
Type[ToolMessage]	A dynamically created Langroid ToolMessage subclass representing the
Type[ToolMessage]	requested tool.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def get_tool_async(
    server: FastMCPServerSpec,
    tool_name: str,
    **client_kwargs: Any,
) -> Type[ToolMessage]:
    """Get a single Langroid ToolMessage subclass for a specific MCP tool name (async).

    This is a convenience wrapper that creates a temporary FastMCPClient.

    Args:
        server: Specification of the FastMCP server to connect to.
        tool_name: The name of the tool to retrieve.
        **client_kwargs: Additional keyword arguments to pass to the
            FastMCPClient constructor (e.g., sampling_handler, roots).

    Returns:
        A dynamically created Langroid ToolMessage subclass representing the
        requested tool.
    """
    async with FastMCPClient(server, **client_kwargs) as client:
        return await client.get_tool_async(tool_name)

get_tool(server, tool_name, **client_kwargs)

Get a single Langroid ToolMessage subclass for a specific MCP tool name (synchronous).

This is a convenience wrapper that creates a temporary FastMCPClient and runs the async get_tool_async function using asyncio.run().

Parameters:

Name	Type	Description	Default
server	FastMCPServerSpec	Specification of the FastMCP server to connect to.	required
tool_name	str	The name of the tool to retrieve.	required
**client_kwargs	Any	Additional keyword arguments to pass to the FastMCPClient constructor (e.g., sampling_handler, roots).	{}

Returns:

Type	Description
Type[ToolMessage]	A dynamically created Langroid ToolMessage subclass representing the
Type[ToolMessage]	requested tool.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

def get_tool(
    server: FastMCPServerSpec,
    tool_name: str,
    **client_kwargs: Any,
) -> Type[ToolMessage]:
    """Get a single Langroid ToolMessage subclass
    for a specific MCP tool name (synchronous).

    This is a convenience wrapper that creates a temporary FastMCPClient and runs the
    async `get_tool_async` function using `asyncio.run()`.

    Args:
        server: Specification of the FastMCP server to connect to.
        tool_name: The name of the tool to retrieve.
        **client_kwargs: Additional keyword arguments to pass to the
            FastMCPClient constructor (e.g., sampling_handler, roots).

    Returns:
        A dynamically created Langroid ToolMessage subclass representing the
        requested tool.
    """
    return asyncio.run(get_tool_async(server, tool_name, **client_kwargs))

get_tools_async(server, **client_kwargs) async

Get all available tools as Langroid ToolMessage subclasses (async).

This is a convenience wrapper that creates a temporary FastMCPClient.

Parameters:

Name	Type	Description	Default
server	FastMCPServerSpec	Specification of the FastMCP server to connect to.	required
**client_kwargs	Any	Additional keyword arguments to pass to the FastMCPClient constructor (e.g., sampling_handler, roots).	{}

Returns:

Type	Description
List[Type[ToolMessage]]	A list of dynamically created Langroid ToolMessage subclasses
List[Type[ToolMessage]]	representing all available tools on the server.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def get_tools_async(
    server: FastMCPServerSpec,
    **client_kwargs: Any,
) -> List[Type[ToolMessage]]:
    """Get all available tools as Langroid ToolMessage subclasses (async).

    This is a convenience wrapper that creates a temporary FastMCPClient.

    Args:
        server: Specification of the FastMCP server to connect to.
        **client_kwargs: Additional keyword arguments to pass to the
            FastMCPClient constructor (e.g., sampling_handler, roots).

    Returns:
        A list of dynamically created Langroid ToolMessage subclasses
        representing all available tools on the server.
    """
    async with FastMCPClient(server, **client_kwargs) as client:
        return await client.get_tools_async()

get_tools(server, **client_kwargs)

Get all available tools as Langroid ToolMessage subclasses (synchronous).

This is a convenience wrapper that creates a temporary FastMCPClient and runs the async get_tools_async function using asyncio.run().

Parameters:

Name	Type	Description	Default
server	FastMCPServerSpec	Specification of the FastMCP server to connect to.	required
**client_kwargs	Any	Additional keyword arguments to pass to the FastMCPClient constructor (e.g., sampling_handler, roots).	{}

Returns:

Type	Description
List[Type[ToolMessage]]	A list of dynamically created Langroid ToolMessage subclasses
List[Type[ToolMessage]]	representing all available tools on the server.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

def get_tools(
    server: FastMCPServerSpec,
    **client_kwargs: Any,
) -> List[Type[ToolMessage]]:
    """Get all available tools as Langroid ToolMessage subclasses (synchronous).

    This is a convenience wrapper that creates a temporary FastMCPClient and runs the
    async `get_tools_async` function using `asyncio.run()`.

    Args:
        server: Specification of the FastMCP server to connect to.
        **client_kwargs: Additional keyword arguments to pass to the
            FastMCPClient constructor (e.g., sampling_handler, roots).

    Returns:
        A list of dynamically created Langroid ToolMessage subclasses
        representing all available tools on the server.
    """
    return asyncio.run(get_tools_async(server, **client_kwargs))

get_mcp_tool_async(server, name, **client_kwargs) async

Get the raw MCP Tool object for a specific tool name (async).

This is a convenience wrapper that creates a temporary FastMCPClient to retrieve the tool definition from the server.

Parameters:

Name	Type	Description	Default
server	FastMCPServerSpec	Specification of the FastMCP server to connect to.	required
name	str	The name of the tool to look up.	required
**client_kwargs	Any	Additional keyword arguments to pass to the FastMCPClient constructor.	{}

Returns:

Type	Description
Optional[Tool]	The raw mcp.types.Tool object from the server, or None if the tool
Optional[Tool]	is not found.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def get_mcp_tool_async(
    server: FastMCPServerSpec,
    name: str,
    **client_kwargs: Any,
) -> Optional[Tool]:
    """Get the raw MCP Tool object for a specific tool name (async).

    This is a convenience wrapper that creates a temporary FastMCPClient to
    retrieve the tool definition from the server.

    Args:
        server: Specification of the FastMCP server to connect to.
        name: The name of the tool to look up.
        **client_kwargs: Additional keyword arguments to pass to the
            FastMCPClient constructor.

    Returns:
        The raw `mcp.types.Tool` object from the server, or `None` if the tool
        is not found.
    """
    async with FastMCPClient(server, **client_kwargs) as client:
        return await client.get_mcp_tool_async(name)

get_mcp_tools_async(server, **client_kwargs) async

Get all available raw MCP Tool objects from the server (async).

This is a convenience wrapper that creates a temporary FastMCPClient to retrieve the list of tool definitions from the server.

Parameters:

Name	Type	Description	Default
server	FastMCPServerSpec	Specification of the FastMCP server to connect to.	required
**client_kwargs	Any	Additional keyword arguments to pass to the FastMCPClient constructor.	{}

Returns:

Type	Description
List[Tool]	A list of raw mcp.types.Tool objects available on the server.

Source code in langroid/agent/tools/mcp/fastmcp_client.py

async def get_mcp_tools_async(
    server: FastMCPServerSpec,
    **client_kwargs: Any,
) -> List[Tool]:
    """Get all available raw MCP Tool objects from the server (async).

    This is a convenience wrapper that creates a temporary FastMCPClient to
    retrieve the list of tool definitions from the server.

    Args:
        server: Specification of the FastMCP server to connect to.
        **client_kwargs: Additional keyword arguments to pass to the
            FastMCPClient constructor.

    Returns:
        A list of raw `mcp.types.Tool` objects available on the server.
    """
    async with FastMCPClient(server, **client_kwargs) as client:
        if not client.client:
            raise RuntimeError("Client not initialized. Use async with FastMCPClient.")
        return await client.client.list_tools()