mcp

langroid/agent/tools/mcp/init.py

FastMCPClient(server, persist_connection=False, forward_images=True, forward_text_resources=False, forward_blob_resources=False, sampling_handler=None, roots=None, log_handler=None, message_handler=None, read_timeout_seconds=None, tool_name_prefix=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
tool_name_prefix	str | None	Optional namespace prepended to generated Langroid tool names as <prefix>__<server-tool-name>. Calls to the MCP server continue to use its original tool name.	None

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

def __init__(
    self,
    server: FastMCPServerSpec,
    persist_connection: bool = False,
    forward_images: bool = True,
    forward_text_resources: bool = False,
    forward_blob_resources: bool = False,
    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,
    tool_name_prefix: str | None = None,
) -> None:
    """Initialize the FastMCPClient.

    Args:
        server: FastMCP server or path to such a server
        tool_name_prefix: Optional namespace prepended to generated Langroid
            tool names as ``<prefix>__<server-tool-name>``. Calls to the MCP
            server continue to use its original tool name.
    """
    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
    # Default a slightly larger read timeout for stdio transports on first
    # connects. Allows flaky subprocess servers a bit more time to boot.
    if read_timeout_seconds is None:
        try:
            default_secs = int(os.getenv("LANGROID_MCP_READ_TIMEOUT", "15"))
            self.read_timeout_seconds = datetime.timedelta(seconds=default_secs)
        except Exception:
            self.read_timeout_seconds = None
    else:
        self.read_timeout_seconds = read_timeout_seconds
    if tool_name_prefix is not None and (
        not isinstance(tool_name_prefix, str)
        or re.fullmatch(r"[a-zA-Z0-9_]+", tool_name_prefix) is None
    ):
        raise ValueError(
            "tool_name_prefix must fully match the ASCII pattern " "[a-zA-Z0-9_]+"
        )
    self.persist_connection = persist_connection
    self.forward_text_resources = forward_text_resources
    self.forward_blob_resources = forward_blob_resources
    self.forward_images = forward_images
    self.tool_name_prefix = tool_name_prefix

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:
        if self.persist_connection:
            await self.connect()
            assert self.client
        else:
            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}")
    return self.tool_model_from_mcp_tool(target)

tool_model_from_mcp_tool(target)

Build a Langroid ToolMessage subclass from an already-fetched MCP Tool object.

This is a pure, synchronous, network-free conversion: it performs no list_tools() round-trip. Pair it with a single list_tools() call to build many tools without re-listing the server once per tool (see :meth:get_tools_async).

Parameters:

Name	Type	Description	Default
target	Tool	The raw mcp.types.Tool (name, description, inputSchema).	required

Returns:

Type	Description
Type[ToolMessage]	A dynamically created Langroid ToolMessage subclass for target.

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

def tool_model_from_mcp_tool(self, target: Tool) -> Type[ToolMessage]:
    """
    Build a Langroid ToolMessage subclass from an already-fetched MCP
    `Tool` object.

    This is a pure, synchronous, network-free conversion: it performs no
    ``list_tools()`` round-trip. Pair it with a single ``list_tools()`` call
    to build many tools without re-listing the server once per tool
    (see :meth:`get_tools_async`).

    Args:
        target: The raw ``mcp.types.Tool`` (name, description, inputSchema).

    Returns:
        A dynamically created Langroid ToolMessage subclass for `target`.
    """
    mcp_tool_name = getattr(target, "name", None)
    if not isinstance(mcp_tool_name, str) or mcp_tool_name == "":
        raise ValueError(f"Invalid MCP tool name for tool {target!r}")
    tool_name = (
        f"{self.tool_name_prefix}__{mcp_tool_name}"
        if self.tool_name_prefix is not None
        else mcp_tool_name
    )

    input_schema = getattr(target, "inputSchema", None)
    schema = input_schema if isinstance(input_schema, dict) else {}
    # The whole input schema, kept because the loop below shadows
    # `schema` with each property and because `$ref` pointers are
    # written against this document, not just against its `$defs`.
    root_schema: Dict[str, Any] = schema
    props = schema.get("properties") or {}
    if not isinstance(props, dict):
        props = {}
    # Registry of shared subschemas ($defs), used to resolve $ref nodes such
    # as pydantic-model params. Captured before the loop below shadows
    # `schema` with each property's schema.
    defs = schema.get("$defs")
    if not isinstance(defs, dict):
        defs = {}
    # Cache of resolved $ref types, shared across all properties so a def
    # referenced by several params is built once and reused.
    ref_cache: Dict[str, Any] = {}
    # Get the list of required fields from JSON Schema
    required_fields = schema.get("required") or []
    if not isinstance(required_fields, list):
        required_fields = []
    required_field_names = {
        name for name in required_fields if isinstance(name, str)
    }
    fields: Dict[str, Tuple[type, Any]] = {}
    for fname, schema in props.items():
        ftype, fld = self._schema_to_field(
            fname,
            schema,
            tool_name,
            is_required=fname in required_field_names,
            defs=defs,
            ref_cache=ref_cache,
        )
        fields[fname] = (ftype, fld)

    # Convert target.name to CamelCase and add Tool suffix
    parts = tool_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, tool_name),
            purpose=(
                str,
                getattr(target, "description", None) or f"Use the tool {tool_name}",
            ),
            __base__=ToolMessage,
            **fields,
        ),
    )
    # Store ALL client configuration needed to recreate a client
    client_config = {
        # Always store a SERVER FACTORY to ensure a fresh transport per call
        "server": self._as_server_factory(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]
    # Nullability is keyed by the model's field names, so apply the same
    # reserved-name renaming the fields went through above.
    nullable_props = _nullable_field_names(props, root_schema)
    tool_model._mcp_nullable_fields = {  # type: ignore[attr-defined]
        renamed.get(name, name) for name in nullable_props
    }
    tool_model._mcp_tool_name = mcp_tool_name  # type: ignore[attr-defined]

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

        # pack up the payload
        # Get exclude fields from model config with proper type checking
        exclude_fields = set()
        model_config = getattr(itself, "model_config", {})
        if (
            isinstance(model_config, dict)
            and "json_schema_extra" in model_config
            and model_config["json_schema_extra"] is not None
            and isinstance(model_config["json_schema_extra"], dict)
            and "exclude" in model_config["json_schema_extra"]
        ):
            exclude_list = model_config["json_schema_extra"]["exclude"]
            if isinstance(exclude_list, (list, set, tuple)):
                exclude_fields = set(exclude_list)

        # Add standard excluded fields
        exclude_fields.update(["request", "purpose"])

        # Drop the None values the server cannot accept (unset Optional
        # placeholders, and strict-mode nulls for non-nullable params), but
        # forward an explicit null for a parameter the server declares
        # nullable -- that is a real argument, not a missing one.
        exclusions = _none_exclusions(itself)
        exclusions.update({name: True for name in exclude_fields})
        payload = itself.model_dump(exclude=exclusions)

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

        client_cfg = getattr(  # type: ignore
            itself.__class__, "_client_config", None
        )
        if not client_cfg:
            # Fallback or error - ideally _client_config should always exist
            raise RuntimeError(f"Client config missing on {itself.__class__}")

        # Connect the client if not yet connected and keep the connection open
        if self.persist_connection:
            if not self.client:
                await self.connect()

            return await self.call_mcp_tool(
                itself.__class__._mcp_tool_name,  # type: ignore[attr-defined]
                payload,
            )

        # open a fresh client, call the tool, then close
        async with FastMCPClient(**client_cfg) as client:  # type: ignore
            return await client.call_mcp_tool(
                itself.__class__._mcp_tool_name,  # type: ignore[attr-defined]
                payload,
            )

    tool_model.call_tool_async = call_tool_async  # type: ignore

    if not hasattr(tool_model, "handle_async"):
        # 3) define handle_async() method with optional agent parameter
        from typing import Union

        async def handle_async(
            self: ToolMessage, agent: Optional[Agent] = None
        ) -> Union[str, Optional[ChatDocument]]:
            """
            Auto-generated handler for MCP tool. Returns ChatDocument with files
            if files are present and agent is provided, otherwise returns text.

            To override: define your own handle_async method with matching signature
            if you need file handling, or simpler signature if you only need text.
            """
            response = await self.call_tool_async()  # type: ignore[attr-defined]
            if response is None:
                return None

            content, files = response

            # If we have files and an agent is provided, return a ChatDocument
            if files and agent is not None:
                return agent.create_agent_response(
                    content=content,
                    files=files,
                )
            else:
                # Otherwise, just return the text content
                return str(content) if content is not None else None

        # 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:
        if self.persist_connection:
            await self.connect()
            assert self.client
        else:
            raise RuntimeError(
                "Client not initialized. Use async with FastMCPClient."
            )
    resp = await self.client.list_tools()
    return [self.tool_model_from_mcp_tool(t) 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:
        if self.persist_connection:
            await self.connect()
            assert self.client
        else:
            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
Optional[tuple[str, list[FileAttachment]]]	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]
) -> Optional[tuple[str, list[FileAttachment]]]:
    """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:
        if self.persist_connection:
            await self.connect()
            assert self.client
        else:
            raise RuntimeError(
                "Client not initialized. Use async with FastMCPClient."
            )
    # Prefer validated call; if server fails to provide structured content
    # despite declaring a schema, fall back to a raw request to bypass
    # client-side validation and still surface the data.
    try:
        result: CallToolResult = await self.client.session.call_tool(
            tool_name,
            arguments,
        )
    except RuntimeError as e:
        msg = str(e)
        if "has an output schema but did not return structured content" not in msg:
            raise
        from mcp.types import (
            CallToolRequest,
            CallToolRequestParams,
            ClientRequest,
        )
        from mcp.types import (
            CallToolResult as _CallToolResult,
        )

        result = await self.client.session.send_request(  # type: ignore[assignment]
            ClientRequest(
                CallToolRequest(
                    params=CallToolRequestParams(
                        name=tool_name, arguments=arguments
                    )
                )
            ),
            _CallToolResult,
        )
    results = self._convert_tool_result(tool_name, result)

    if isinstance(results, str):
        return results, []

    return results

mcp_tool(server, tool_name)

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

Usage

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

The server may be a string/URL/FastMCP/ClientTransport, or a zero-arg callable returning one of those, e.g. lambda: StdioTransport(...). Using a factory ensures a fresh transport per connection under fastmcp>=2.13.

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

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

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

    The `server` may be a string/URL/FastMCP/ClientTransport, or a zero-arg
    callable returning one of those, e.g. `lambda: StdioTransport(...)`. Using a
    factory ensures a fresh transport per connection under fastmcp>=2.13.
    """

    def decorator(user_cls: Type[ToolMessage]) -> Type[ToolMessage]:
        # build the “real” ToolMessage subclass for this server/tool
        RealTool: Type[ToolMessage] = get_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

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_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_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_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_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()