Custom Tools

Custom Tools

This guide covers how to build tools with advanced features like caching, authorization, structured output, and cancellation.

Basic Tool

from agiwo.tool import BaseTool, ToolContext, ToolResult


class GreetTool(BaseTool):
    name = "greet"
    description = "Greet a person by name"

    def get_parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "Person's name"},
            },
            "required": ["name"],
        }

    async def execute(
        self,
        parameters: dict,
        context: ToolContext,
        abort_signal=None,
    ) -> ToolResult:
        name = parameters["name"]
        return ToolResult.success(
            tool_name=self.name,
            content=f"Hello, {name}!",
            content_for_user=f"Greeted {name}",
        )

Handling Errors

Use ToolResult.failed() for expected errors:

async def execute(self, parameters, context, abort_signal) -> ToolResult:
    try:
        result = await self._do_work(parameters)
        return ToolResult.success(tool_name=self.name, content=result)
    except ExternalAPIError as error:
        return ToolResult.failed(
            tool_name=self.name,
            error=f"API error: {error}",
        )

Structured Output

Pass structured data via the output field:

return ToolResult.success(
    tool_name=self.name,
    content=f"Found {len(results)} results",
    output={"results": results, "count": len(results)},
)

Caching

Enable session-scoped caching for expensive operations:

class EmbeddingTool(BaseTool):
    cacheable = True
    timeout_seconds = 60

Abort Signals

Check abort_signal in long-running tools:

async def execute(self, parameters, context, abort_signal) -> ToolResult:
    for item in large_dataset:
        if abort_signal and abort_signal.is_aborted():
            return ToolResult.aborted(tool_name=self.name)
        await process(item)
    return ToolResult.success(tool_name=self.name, content="Done")

Concurrency Safety

• True means the tool can run concurrently with other tools.
• False means the tool must run alone because it mutates shared state or depends on ordering.

ToolContext

ToolContext provides runtime information such as agent_name and session_id.

Denial Pattern

For permission gating:

async def execute(self, parameters, context, abort_signal) -> ToolResult:
    if not self._check_permission(parameters, context):
        return ToolResult.denied(
            tool_name=self.name,
            reason="Requires admin access",
        )

Registering Tools

Pass tools when creating an agent:

agent = Agent(
    AgentConfig(name="assistant", description="...", system_prompt="..."),
    model=model,
    tools=[GreetTool(), WeatherTool(), CalculatorTool()],
)