mcp

langroid/agent/tools/mcp/init.py

FastMCPClient(server)

A client for interacting with a FastMCP server.

Provides async context manager functionality to safely manage resources.

Parameters:

Name	Type	Description	Default
server	str | FastMCP[Any] | ClientTransport	FastMCP server or path to such a server	required

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

def __init__(self, server: str | FastMCP[Any] | ClientTransport) -> None:
    """Initialize the FastMCPClient.

    Args:
        server: FastMCP server or path to such a server
    """
    self.server = server
    self.client = None
    self._cm = None

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_langroid_tool(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_langroid_tool(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"

    # 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,
        ),
    )
    tool_model._server = self.server  # 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"])
        # open a fresh client, call the tool, then close
        async with FastMCPClient(self.__class__._server) 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_langroid_tools() 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_langroid_tools(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()
    tools: List[Type[ToolMessage]] = []
    for t in resp:
        tools.append(await self.get_langroid_tool(t.name))
    return tools

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)

mcp_tool(server, tool_name)

Decorator: declare a ToolMessage class bound to a FastMCP tool.

Usage

@fastmcp_tool("/path/to/server.py", "get_weather") class WeatherTool: def pretty(self) -> str: return f"Temp is {self.temperature}"

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

def mcp_tool(
    server: str, tool_name: str
) -> Callable[[Type[ToolMessage]], Type[ToolMessage]]:
    """Decorator: declare a ToolMessage class bound to a FastMCP tool.

    Usage:
        @fastmcp_tool("/path/to/server.py", "get_weather")
        class WeatherTool:
            def pretty(self) -> str:
                return f"Temp is {self.temperature}"
    """

    def decorator(user_cls: Type[ToolMessage]) -> Type[ToolMessage]:
        # build the “real” ToolMessage subclass for this server/tool
        RealTool: Type[ToolMessage] = get_langroid_tool(server, tool_name)

        # copy user‐defined methods / attributes onto RealTool
        for name, attr in user_cls.__dict__.items():
            if name.startswith("__") and name.endswith("__"):
                continue
            setattr(RealTool, name, attr)

        # preserve the user’s original name if you like:
        RealTool.__name__ = user_cls.__name__
        return RealTool

    return decorator