Custom Tools

You can extend Mango with any tool you need — an HTTP call, a second database, a calculation engine. The LLM will call your tool exactly like the built-in ones.

Minimal example

from mango.tools.base import Tool, ToolResult
from mango.llm import ToolDef, ToolParam

class GetExchangeRateTool(Tool):

    @property
    def definition(self) -> ToolDef:
        return ToolDef(
            name="get_exchange_rate",
            description="Get the current exchange rate between two currencies.",
            params=[
                ToolParam(name="from_currency", type="string", description="Base currency (e.g. USD)", required=True),
                ToolParam(name="to_currency", type="string", description="Target currency (e.g. EUR)", required=True),
            ],
        )

    async def execute(self, from_currency: str, to_currency: str) -> ToolResult:
        # Your logic here
        rate = fetch_rate(from_currency, to_currency)
        return ToolResult(success=True, data={"rate": rate, "pair": f"{from_currency}/{to_currency}"})

Register it

tools = ToolRegistry()

# Built-in tools
for tool in build_mongo_tools(db, memory):
    tools.register(tool)

# Your custom tool
tools.register(GetExchangeRateTool())

The LLM will automatically see your tool's name and description and call it when relevant.

ToolParam reference

@dataclass
class ToolParam:
    name: str
    type: str           # "string" | "integer" | "number" | "boolean" | "array" | "object"
    description: str
    required: bool = True
    enum: list | None = None     # restrict to specific values
    items: dict | None = None    # for array types, describe the items

ToolResult reference

@dataclass
class ToolResult:
    success: bool
    data: Any            # any JSON-serializable value
    error: str | None = None

    def as_text(self) -> str:
        """How the result is shown to the LLM."""

Return success=False with a descriptive error message to let the LLM know the tool failed and potentially retry with different arguments.

Async tools

All tools are async. For synchronous operations, just return directly — no need to use await:

async def execute(self, **kwargs) -> ToolResult:
    result = some_sync_function()  # sync is fine
    return ToolResult(success=True, data=result)

For blocking I/O (database calls, HTTP), use asyncio.to_thread:

import asyncio

async def execute(self, collection: str) -> ToolResult:
    docs = await asyncio.to_thread(self._db_call, collection)
    return ToolResult(success=True, data=docs)