ant_ai.tools.tool

ToolError

Bases: Exception

Raised by a tool to report a failure the model can recover from.

Tools signal failure by raising, not by returning an error string: ToolStep turns any exception into an ERROR: ... result for the model and marks the call is_error=True, whereas a returned value is a success -- the step does not inspect content. ToolError names the expected case, where the message is written for the model; unexpected exceptions take the same path.

MCP tools raise it when the server answers with isError.

Source code in src/ant_ai/tools/tool.py

class ToolError(Exception):
    """Raised by a tool to report a failure the model can recover from.

    Tools signal failure by raising, not by returning an error string: `ToolStep`
    turns any exception into an `ERROR: ...` result for the model and marks the
    call `is_error=True`, whereas a returned value is a success -- the step does
    not inspect content. `ToolError` names the expected case, where the message
    is written for the model; unexpected exceptions take the same path.

    MCP tools raise it when the server answers with `isError`.
    """

Tool pydantic-model

Bases: BaseModel

Single public abstraction for tools.

Usage patterns:

1) Subclass for namespaces (methods → tools "ClassName.method"):

class Math(Tool):
    def add(self, x: int, y: int) -> int: ...
    def mul(self, x: int, y: int) -> int: ...

2) Decorate functions with @tool:

@tool
def ping(host: str) -> str: ...

3) MCP tools loaded via mcp_tools_from_url(...):

tools = await mcp_tools_from_url("http://localhost:8000/mcp")
# returns list[Tool] that proxy to MCP tools

Internally, a Tool is either:

• a namespace (class-based, many methods)
• a single-callable Python tool (function-based)
• or a proxy for an MCP tool (created via mcp_tools_from_url)

Notes

When defining a namespace tool, so as a subclass of Tool, the documentation of the class itself is not used by the agent. Instead, it's the documentation of each method that is used.

Show JSON schema:

{
  "description": "Single public abstraction for tools.\n\nUsage patterns:\n\n1) Subclass for *namespaces* (methods \u2192 tools \"ClassName.method\"):\n\n```python\nclass Math(Tool):\n    def add(self, x: int, y: int) -> int: ...\n    def mul(self, x: int, y: int) -> int: ...\n```\n\n2) Decorate functions with @tool:\n\n```python\n@tool\ndef ping(host: str) -> str: ...\n```\n\n3) MCP tools loaded via `mcp_tools_from_url(...)`:\n\n```python\ntools = await mcp_tools_from_url(\"http://localhost:8000/mcp\")\n# returns list[Tool] that proxy to MCP tools\n```\n\nInternally, a Tool is either:\n\n- a *namespace* (class-based, many methods)\n- a *single-callable* Python tool (function-based)\n- or a proxy for an MCP tool (created via `mcp_tools_from_url`)\n\nNotes:\n    When defining a namespace tool, so as a subclass of Tool, the documentation of the class itself is not used by the agent. Instead, it's the documentation of each method that is used.",
  "properties": {
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Tool name.",
      "title": "Name"
    },
    "description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Tool description. Is used by the LLM to decide whether to call or not the specific tool.",
      "title": "Description"
    },
    "parameters": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The parameters needed by the tool. This is a self-constructed field.",
      "title": "Parameters"
    }
  },
  "title": "Tool",
  "type": "object"
}

Config:

• arbitrary_types_allowed: True

Fields:

• name (str | None)
• description (str | None)
• parameters (dict[str, Any] | None)
• __namespace_methods__ (list[str])

Validators:

• _set_defaults

Source code in src/ant_ai/tools/tool.py

class Tool(BaseModel):
    """
    Single public abstraction for tools.

    Usage patterns:

    1) Subclass for *namespaces* (methods → tools "ClassName.method"):

    ```python
    class Math(Tool):
        def add(self, x: int, y: int) -> int: ...
        def mul(self, x: int, y: int) -> int: ...
    ```

    2) Decorate functions with @tool:

    ```python
    @tool
    def ping(host: str) -> str: ...
    ```

    3) MCP tools loaded via `mcp_tools_from_url(...)`:

    ```python
    tools = await mcp_tools_from_url("http://localhost:8000/mcp")
    # returns list[Tool] that proxy to MCP tools
    ```

    Internally, a Tool is either:

    - a *namespace* (class-based, many methods)
    - a *single-callable* Python tool (function-based)
    - or a proxy for an MCP tool (created via `mcp_tools_from_url`)

    Notes:
        When defining a namespace tool, so as a subclass of Tool, the documentation of the class itself is not used by the agent. Instead, it's the documentation of each method that is used.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: str | None = Field(default=None, description="Tool name.")
    description: str | None = Field(
        default=None,
        description="Tool description. Is used by the LLM to decide whether to call or not the specific tool.",
    )
    parameters: dict[str, Any] | None = Field(
        default=None,
        description="The parameters needed by the tool. This is a self-constructed field.",
    )

    _func: Callable[..., Any] | None = PrivateAttr(default=None)
    __namespace_methods__: list[str] = []
    _func_signature: inspect.Signature | None = PrivateAttr(default=None)
    _wants_ctx: bool = PrivateAttr(default=False)

    @model_validator(mode="after")
    def _set_defaults(self) -> Tool:
        if self.name is None:
            self.name: str = self.__class__.__name__

        if self.description is None:
            doc: str = inspect.getdoc(self.__class__) or ""
            self.description = doc.strip() or ""

        if self.parameters is None:
            self.parameters: dict[str, Any] = {"type": "object", "properties": {}}

        return self

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """
        When you subclass Tool, this inspects the class body and collects
        all public callables as "namespace methods".

        Namespace methods declared on an intermediate Tool-subclass ancestor
        (e.g. a shared base class's own tool methods, like `MemoryTool.search`)
        are inherited even when a concrete subclass doesn't redefine them —
        only methods first introduced on `Tool`/`BaseModel` itself, or on a
        non-Tool mixin (e.g. an abstract protocol's own methods), are excluded.
        """
        super().__init_subclass__(**kwargs)
        inherited_names: set[str] = {
            name for klass in cls.__mro__[1:] for name in klass.__dict__
        }
        own_methods: list[str] = [
            name
            for name, value in cls.__dict__.items()
            if _is_public_callable(name, value) and name not in inherited_names
        ]
        inherited_namespace_methods: list[str] = []
        for klass in cls.__mro__[1:]:
            existing = klass.__dict__.get("__namespace_methods__")
            if existing:
                inherited_namespace_methods = list(existing)
                break
        cls.__namespace_methods__: list[str] = list(
            dict.fromkeys(own_methods + inherited_namespace_methods)
        )

    def _call_func(self, *args: Any, **kwargs: Any) -> Any:
        """
        Internal: call self._func with positional + keyword args,
        normalized via the underlying function's signature.
        """
        if self._func is None:
            raise RuntimeError("Cannot invoke a namespace Tool directly.")

        # Lazily cache the signature
        if self._func_signature is None:
            self._func_signature = inspect.signature(self._func)

        bound = self._func_signature.bind_partial(*args, **kwargs)
        return self._func(*bound.args, **bound.kwargs)

    @property
    def is_namespace(self) -> bool:
        """
        True if this Tool is a namespace (class with methods),
        False if it's a single-callable tool created by @tool or MCP.
        """
        return (
            bool(getattr(self.__class__, "__namespace_methods__", []))
            and self._func is None
        )

    @property
    def wants_context(self) -> bool:
        """
        True if this tool's underlying function declares a `ctx` parameter.

        `ToolStep` injects the real `InvocationContext` for such tools at
        call time; `ctx` is never exposed in the LLM-facing JSON schema.
        """
        return self._wants_ctx

    @classmethod
    def _from_function(
        cls,
        func: Callable[..., Any],
        *,
        name: str | None = None,
        description: str | None = None,
        args_model: type[BaseModel] | None = None,
    ) -> Tool:
        tool_name: str = name or getattr(func, "__name__", None) or ""

        if args_model is None:
            args_model = _build_args_model_from_signature(func, f"{tool_name}_Args")

        schema = args_model.model_json_schema()
        schema.setdefault("type", "object")

        inst = cls(
            name=tool_name,
            description=(description or inspect.getdoc(func) or "").strip(),
            parameters=schema,
        )
        inst._func = func
        inst._wants_ctx = "ctx" in inspect.signature(func).parameters
        return inst

    @classmethod
    def _from_mcp_descriptor(
        cls,
        *,
        url: str,
        mcp_tool: Any,
        transport: Literal["http", "sse"] = "http",
        headers: dict[str, str] | None = None,
        namespace: str | None = None,
        unwrap_result: bool = True,
    ) -> Tool:
        full_name: str = f"{namespace}.{mcp_tool.name}" if namespace else mcp_tool.name
        params: dict = dict(mcp_tool.input_schema or {})
        params.setdefault("type", "object")

        async def _call_mcp(**kwargs: Any) -> Any:
            # Open a short-lived MCP connection per call.
            # This keeps the Tool self-contained and agent-friendly.
            if transport == "sse":
                ctx_manager = sse_client(url, headers=headers)
            else:
                ctx_manager = streamable_http_client(url)
            async with ctx_manager as client_context:
                read, write = client_context[0], client_context[1]
                async with MCPClientSession(read, write) as session:
                    await session.initialize()
                    result: mcp.types.CallToolResult = await session.call_tool(
                        mcp_tool.name, arguments=kwargs
                    )

            if not unwrap_result:
                return result

            first = result.content[0] if result.content else None
            text: str | None = (
                first.text if isinstance(first, mcp.types.TextContent) else None
            )
            if result.is_error:
                raise ToolError(text or f"MCP tool '{mcp_tool.name}' failed")

            if result.structured_content not in (None, {}):
                return result.structured_content

            if text is not None:
                return text

            return result

        inst = cls(
            name=full_name, description=mcp_tool.description or "", parameters=params
        )
        inst._func = _call_mcp
        return inst

    def _expand_namespace(self) -> list[Tool]:
        """
        For a namespace Tool (class-based), build a list of
        single-callable Tools named "ClassName.method".
        """
        tools: list[Tool] = []
        ns: str = self.name or self.__class__.__name__

        for method_name in self.__class__.__namespace_methods__:
            bound = getattr(self, method_name)
            doc = (
                inspect.getdoc(bound)
                or inspect.getdoc(getattr(self.__class__, method_name))
                or ""
            )

            method_tool: Tool = Tool._from_function(
                func=bound,
                name=f"{ns}_{method_name}",
                description=doc.strip() or f"Method {method_name} on {ns}",
            )
            tools.append(method_tool)

        return tools

    def invoke(self, *args, **kwargs: Any) -> Any:
        """
        Synchronous invocation.

        For plain Python tools, this returns the result directly.
        For MCP-backed tools, this returns a coroutine; you should use
        `await tool.ainvoke(...)` in async code.
        """
        if self._func is None:
            raise RuntimeError("Cannot invoke a namespace Tool directly.")
        return self._call_func(*args, **kwargs)

    async def ainvoke(self, *args, **kwargs: Any) -> Any:
        """
        Asynchronous invocation.

        - If the underlying tool is async → await it directly.
        - If it's sync → run it in a thread executor so we don't block the loop.
        """
        if self._func is None:
            raise RuntimeError("Cannot invoke a namespace Tool directly.")

        if inspect.iscoroutinefunction(self._func):
            # Call it directly; don't go through invoke() to avoid double checks
            result: Any = self._call_func(*args, **kwargs)
            # result should already be a coroutine
            return await cast(Awaitable[Any], result)

        loop: AbstractEventLoop = asyncio.get_running_loop()
        bound: partial[Any] = functools.partial(self._call_func, *args, **kwargs)
        return await loop.run_in_executor(None, bound)

    @model_serializer
    def _serialize(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description or "",
                "parameters": self.parameters or {"type": "object", "properties": {}},
            },
        }

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        if self._func is None:
            raise RuntimeError("Cannot call a namespace Tool directly.")
        return self._call_func(*args, **kwargs)

name pydantic-field

name: str | None = None

Tool name.

description pydantic-field

description: str | None = None

Tool description. Is used by the LLM to decide whether to call or not the specific tool.

parameters pydantic-field

parameters: dict[str, Any] | None = None

The parameters needed by the tool. This is a self-constructed field.

is_namespace property

is_namespace: bool

True if this Tool is a namespace (class with methods), False if it's a single-callable tool created by @tool or MCP.

wants_context property

wants_context: bool

True if this tool's underlying function declares a ctx parameter.

ToolStep injects the real InvocationContext for such tools at call time; ctx is never exposed in the LLM-facing JSON schema.

__init_subclass__

__init_subclass__(**kwargs: Any) -> None

When you subclass Tool, this inspects the class body and collects all public callables as "namespace methods".

Namespace methods declared on an intermediate Tool-subclass ancestor (e.g. a shared base class's own tool methods, like MemoryTool.search) are inherited even when a concrete subclass doesn't redefine them — only methods first introduced on Tool/BaseModel itself, or on a non-Tool mixin (e.g. an abstract protocol's own methods), are excluded.

Source code in src/ant_ai/tools/tool.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """
    When you subclass Tool, this inspects the class body and collects
    all public callables as "namespace methods".

    Namespace methods declared on an intermediate Tool-subclass ancestor
    (e.g. a shared base class's own tool methods, like `MemoryTool.search`)
    are inherited even when a concrete subclass doesn't redefine them —
    only methods first introduced on `Tool`/`BaseModel` itself, or on a
    non-Tool mixin (e.g. an abstract protocol's own methods), are excluded.
    """
    super().__init_subclass__(**kwargs)
    inherited_names: set[str] = {
        name for klass in cls.__mro__[1:] for name in klass.__dict__
    }
    own_methods: list[str] = [
        name
        for name, value in cls.__dict__.items()
        if _is_public_callable(name, value) and name not in inherited_names
    ]
    inherited_namespace_methods: list[str] = []
    for klass in cls.__mro__[1:]:
        existing = klass.__dict__.get("__namespace_methods__")
        if existing:
            inherited_namespace_methods = list(existing)
            break
    cls.__namespace_methods__: list[str] = list(
        dict.fromkeys(own_methods + inherited_namespace_methods)
    )

invoke

invoke(*args, **kwargs: Any) -> Any

Synchronous invocation.

For plain Python tools, this returns the result directly. For MCP-backed tools, this returns a coroutine; you should use await tool.ainvoke(...) in async code.

Source code in src/ant_ai/tools/tool.py

def invoke(self, *args, **kwargs: Any) -> Any:
    """
    Synchronous invocation.

    For plain Python tools, this returns the result directly.
    For MCP-backed tools, this returns a coroutine; you should use
    `await tool.ainvoke(...)` in async code.
    """
    if self._func is None:
        raise RuntimeError("Cannot invoke a namespace Tool directly.")
    return self._call_func(*args, **kwargs)

ainvoke async

ainvoke(*args, **kwargs: Any) -> Any

Asynchronous invocation.

• If the underlying tool is async → await it directly.
• If it's sync → run it in a thread executor so we don't block the loop.

Source code in src/ant_ai/tools/tool.py

async def ainvoke(self, *args, **kwargs: Any) -> Any:
    """
    Asynchronous invocation.

    - If the underlying tool is async → await it directly.
    - If it's sync → run it in a thread executor so we don't block the loop.
    """
    if self._func is None:
        raise RuntimeError("Cannot invoke a namespace Tool directly.")

    if inspect.iscoroutinefunction(self._func):
        # Call it directly; don't go through invoke() to avoid double checks
        result: Any = self._call_func(*args, **kwargs)
        # result should already be a coroutine
        return await cast(Awaitable[Any], result)

    loop: AbstractEventLoop = asyncio.get_running_loop()
    bound: partial[Any] = functools.partial(self._call_func, *args, **kwargs)
    return await loop.run_in_executor(None, bound)

tool

tool(_func: Callable[..., Any]) -> Tool

tool(
    _func: None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    args_model: type[BaseModel] | None = None,
) -> Callable[[Callable[..., Any]], Tool]

tool(
    _func: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    args_model: type[BaseModel] | None = None,
) -> Callable[[Callable[..., Any]], Tool] | Tool

Decorator for functions that turns them into Tool instances.

From the user's POV:

@tool
def ping(host: str) -> str: ...

Source code in src/ant_ai/tools/tool.py

def tool(
    _func: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    args_model: type[BaseModel] | None = None,
) -> Callable[[Callable[..., Any]], Tool] | Tool:
    """
    Decorator for *functions* that turns them into Tool instances.

    From the user's POV:

        @tool
        def ping(host: str) -> str: ...
    """

    def decorator(func: Callable[..., Any]) -> Tool:
        return Tool._from_function(
            func,
            name=name,
            description=description,
            args_model=args_model,
        )

    if _func is not None:
        return decorator(_func)

    return decorator

mcp_tools_from_url async

mcp_tools_from_url(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    transport: Literal["http", "sse"] = "http",
    namespace: str | None = None,
    unwrap_result: bool = True,
) -> list[Tool]

Connect to an MCP server and adapt its tools into Tool objects.

• url: MCP server endpoint, e.g. "http://localhost:8000/mcp"
• headers: Optional HTTP headers (useful for authentication).
• transport: "http" (default, streamable HTTP) or "sse" (Server-Sent Events).
• namespace: optional prefix for tool names, e.g. "weather.get_forecast"
• unwrap_result:
  • if True: return structuredContent or first text fragment
  • if False: return the full CallToolResult

Example:

tools = await mcp_tools_from_url("http://localhost:8000/mcp", namespace="remote")
agent = Agent(tools=tools, ...)  # your agent just sees Tool objects

Source code in src/ant_ai/tools/tool.py

async def mcp_tools_from_url(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    transport: Literal["http", "sse"] = "http",
    namespace: str | None = None,
    unwrap_result: bool = True,
) -> list[Tool]:
    """
    Connect to an MCP server and adapt its tools into `Tool` objects.

    - `url`: MCP server endpoint, e.g. "http://localhost:8000/mcp"
    - `headers`: Optional HTTP headers (useful for authentication).
    - `transport`: `"http"` (default, streamable HTTP) or `"sse"` (Server-Sent Events).
    - `namespace`: optional prefix for tool names, e.g. "weather.get_forecast"
    - `unwrap_result`:
        * if True: return structuredContent or first text fragment
        * if False: return the full `CallToolResult`

    Example:

        tools = await mcp_tools_from_url("http://localhost:8000/mcp", namespace="remote")
        agent = Agent(tools=tools, ...)  # your agent just sees Tool objects
    """
    if transport == "sse":
        ctx_manager = sse_client(url, headers=headers)
    else:
        ctx_manager = streamable_http_client(url)

    async with ctx_manager as client_context:
        read, write = client_context[0], client_context[1]
        async with MCPClientSession(read, write) as session:
            await session.initialize()
            tools_resp: mcp.types.ListToolsResult = await session.list_tools()
            mcp_tools: list[mcp.types.Tool] = tools_resp.tools

    tools: list[Tool] = [
        Tool._from_mcp_descriptor(
            url=url,
            mcp_tool=mcp_tool,
            transport=transport,
            headers=headers,
            namespace=namespace,
            unwrap_result=unwrap_result,
        )
        for mcp_tool in mcp_tools
    ]

    return tools