Tools

@tool decorator

The easiest way to create a tool is the @tool decorator. It wraps any Python function (sync or async) and derives the tool name from the function name and the description from the docstring.

from cyclops.toolkit import tool


@tool
def get_weather(location: str) -> str:
    """Get the current weather for a location."""
    return f"Sunny, 72°F in {location}"

You can also use @tool() with parentheses:

@tool()
def get_weather(location: str) -> str:
    """Get the current weather for a location."""
    return f"Sunny, 72°F in {location}"

Custom names and descriptions

Override the function name or docstring by passing keyword arguments:

@tool(name="weather_lookup", description="Fetch live weather data for any city.")
def _internal_get_weather(location: str) -> str:
    return f"Sunny, 72°F in {location}"

This is useful when the function name is not suitable as a public tool name, or when you want a richer description than the docstring provides.

BaseTool subclass

For tools that need to hold state or have complex initialisation, subclass BaseTool and override execute().

from cyclops.toolkit.tool import BaseTool


class DatabaseTool(BaseTool):
    def __init__(self):
        super().__init__(name="query_db", description="Query an in-memory database.")
        self.data = {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}

    async def execute(self, table: str, user_id: int = None) -> str:
        rows = self.data.get(table, [])
        if user_id is not None:
            rows = [r for r in rows if r.get("id") == user_id]
        return str(rows)


db_tool = DatabaseTool()

execute() must be defined with the exact parameter signature you want exposed to the LLM. Cyclops inspects that signature to build the JSON schema.

Sync vs async tools

Both sync functions and async coroutines work. The agent calls sync tools directly, and runs async tools from sync contexts using a nested-loop-safe fallback (ThreadPoolExecutor) when needed. Prefer async for I/O-bound work.

import httpx
from cyclops.toolkit import tool


@tool
async def fetch_url(url: str) -> str:
    """Fetch the text content of a URL."""
    async with httpx.AsyncClient() as client:
        r = await client.get(url, timeout=10)
        return r.text[:500]

Parameter schemas

Cyclops builds the JSON schema of a tool’s parameters with pydantic from the function signature, so any type pydantic understands works: list[str] gets items, Literal and Enum become enum, Optional[T] allows null, pydantic models become nested objects, and default values are kept. Parameters without a default are required.

Parameter descriptions come from a Google or NumPy style docstring, or from Annotated[..., Field(description=...)]. When both describe a parameter, the docstring wins. The docstring text before the parameter section becomes the tool description.

from typing import Annotated, Literal

from pydantic import BaseModel, Field


class Filter(BaseModel):
    field: str
    value: str


@tool
def search(
    query: str,
    filters: list[Filter],
    sort: Literal["relevance", "date"] = "relevance",
    limit: Annotated[int, Field(description="Maximum number of hits.", ge=1)] = 5,
) -> str:
    """Search the index.

    Args:
        query: Text to search for.
        filters: Conditions every hit must match.
        sort: Result order.
    """
    return f"Results for '{query}'"

When a function tool is called, its arguments are validated against the same schema, so filters arrives as a list of Filter instances. Unknown arguments are rejected unless the function takes **kwargs. A BaseTool subclass gets its schema from the signature of execute() and receives the arguments as passed.

Tool results

A tool may return a string, any JSON-serializable value, or a ToolResult. A ToolResult carries content blocks (TextContent, ImageContent), structured data in result, and an error flag.

from cyclops.toolkit import ImageContent, TextContent, ToolResult


@tool
def render_chart(series: list[float]) -> ToolResult:
    """Render a chart of the series."""
    if not series:
        return ToolResult.error_result("series is empty")
    return ToolResult(
        content=[
            TextContent(text=f"Chart of {len(series)} points"),
            ImageContent(data=png_base64(series), mime_type="image/png"),
        ],
        result={"points": len(series)},
    )

The agent sends every result to the model through ToolResult.to_message_content(), wrapping a plain return value with ToolResult.success_result():

• Text renders as a string. result is rendered as JSON only when there are no content blocks, so a plain string stays as it is and a dict or list becomes JSON.
• A failed result starts with Error:, so the model sees that the call failed. A tool that raises, a denied call and invalid arguments are reported the same way.
• Images are sent in the tool message for providers whose tool messages take images through LiteLLM: Anthropic, Bedrock, Gemini, and Gemini models on Vertex AI. Other providers, OpenAI and Vertex AI partner models among them, accept only text there: the tool message gets the text, and the images follow in a user message after the round’s tool messages. Through a router, images stay in the tool message only when every deployment takes them, and with fallback models only when every model does. In naive tool mode the images go in the user message that carries the tool result.
• A provider that rejects a user message right after a tool message (reported for Mistral) cannot take images from tools; return text only for such models.

Hooks (on_tool_end) and RunResult.tool_calls receive the text of the result.

Annotations

Tools can carry behaviour hints that mirror MCP tool annotations. The agent reads read_only to decide which calls may run in parallel (see below); the others are not enforced, and hooks and other code can read all of them from tool.annotations.

from cyclops.toolkit import ToolAnnotations


@tool(annotations=ToolAnnotations(read_only=True, open_world=True))
def fetch_page(url: str) -> str:
    """Fetch a web page."""
    ...

BaseTool subclasses pass annotations= to super().__init__(). The fields are read_only, destructive, idempotent and open_world; each defaults to None (unknown).

Parallel calls and timeouts

When the model asks for several tools in one step, calls to tools marked read_only=True run at the same time. Every other call may change something, so those run one at a time in the order the model gave them, alongside the read-only ones. AgentConfig.max_parallel_tool_calls (default 10) caps how many calls of a step run at once. A call waiting for its turn has not started: it gets no ToolCallStarted event until it runs. If middleware raises, that call ends the step: the calls still running are cancelled, and a call that has not started by then never starts and gets Tool call cancelled.

AgentConfig.tool_timeout sets how many seconds any call may run, and a tool’s own timeout overrides it:

agent = Agent(config=AgentConfig(model="gpt-4o-mini", tool_timeout=30), tools=[...])


@tool(timeout=120)
def build_report(month: str) -> str:
    """Build the monthly report."""
    ...

A call that runs over is cancelled and the model gets Error: <tool> timed out after <n> seconds. Middleware after_tool, on_tool_error and the ToolCallFinished event see it as any other failed call, with an asyncio.TimeoutError. A sync tool runs in a worker thread, which cannot be stopped: the agent moves on, but the function runs until it returns. MCP tools also have the client’s own call_timeout.

In the same way, a tool’s own max_output_chars overrides AgentConfig.max_tool_output_chars, the longest result text the model gets whole; a longer one is stored, and the model gets its start and end and can read the rest. See Tool output limit.

Deferred tools

Every tool’s schema is sent with every completion, which costs context when an agent has many tools. Mark a tool defer_loading=True and it is left out until the model asks for it: the agent then sends a search_tools tool, and each tool a search finds is sent from the next step to the end of the run. Tools not marked are always sent.

@tool(defer_loading=True)
def create_invoice(customer: str, amount: float) -> str:
    """Create an invoice for a customer."""
    ...

BaseTool subclasses pass defer_loading= to super().__init__(), and MCPClient.attach(..., defer_loading=True) defers every tool of that server.

search_tools takes a query of keywords and returns up to ten deferred tools that share words with their name and description (names split at underscores), most shared words first, with tools not loaded yet ahead of loaded ones. The model gets their names and descriptions; their schemas arrive with the next completion, or in the system prompt in naive tool mode. A deferred tool the model calls by name without searching still runs. With any tool deferred, search_tools is a reserved name.

Found tools last for the run. The next run, in a session too, starts again from the tools that are always sent, and the model searches again for what it needs.

search_tools goes through middleware like any other tool. With ToolPermissions(default="deny"), allow it, or no deferred tool can load:

ToolPermissions(default="deny", allow=["read_*", "search_tools"])

A deferred tool that a rule denies is still found, and denied when called.

A search that adds tools changes what is sent before the conversation: the tools, which come first in the prompt, or the system prompt in naive tool mode. The next completion misses the prompt cache and writes it again over the tools, the system prompt and the history; later steps hit it again. The first completion of the next run misses as well, as it goes back to the tools always sent.

The search runs in the agent for every provider, Anthropic included. Anthropic’s and OpenAI’s own tool search keep the tools sent unchanged and so keep the cache; using them is tracked in #113.

ToolRegistry

ToolRegistry is a named collection of tools. It is useful when you want to manage a set of tools separately from any single agent, such as in a plugin or a shared tool library.

from cyclops.toolkit.registry import ToolRegistry
from cyclops.toolkit import tool


registry = ToolRegistry()


@tool(registry=registry)
def ping(host: str) -> str:
    """Ping a host and return the round-trip time."""
    return f"pong from {host} in 12ms"


# Or register a tool manually. Names must be unique; registering
# a second tool with the same name raises ValueError.
@tool
def pong(host: str) -> str:
    """Reply to a ping."""
    return f"pong to {host}"


registry.register(pong)

# Retrieve and inspect:
print(registry.list_tools())          # ['ping', 'pong']
print(registry.get_tool("ping"))      # <Tool ping>
print(registry.get_definitions())     # {'ping': ToolDefinition(...)}

# Execute a tool by name:
import asyncio
result = asyncio.run(registry.execute_tool("ping", host="example.com"))

Using tools with Agent

Pass a list of tools (decorated functions or BaseTool instances) to the Agent constructor. Tool names must be unique; the constructor raises ValueError on a duplicate.

from cyclops import Agent, AgentConfig
from cyclops.toolkit import tool


@tool
def calculate(operation: str, a: float, b: float) -> str:
    """Perform a basic math operation: add, subtract, multiply, divide."""
    ops = {"add": a + b, "subtract": a - b, "multiply": a * b, "divide": a / b}
    return str(ops.get(operation, "unknown operation"))


config = AgentConfig(model="groq/llama-3.1-8b-instant")
agent = Agent(config, tools=[get_weather, db_tool, calculate])

print(agent.run("What is the weather in Tokyo?"))

You can also pull tools out of a registry:

tools = [registry.get_tool(name) for name in registry.list_tools()]
agent = Agent(config, tools=tools)

Note

If the model does not support native function calling (e.g. some Ollama models), set tool_mode="naive" in AgentConfig. With tool_mode="auto", Cyclops falls back to prompt-based tool invocation for that agent when LiteLLM raises UnsupportedParamsError and litellm.get_supported_openai_params does not list tools for the model (through a Router, for every deployment behind the alias). Any other error is raised unchanged.

Full example

examples/agent_with_tools.py

"""Agent with tools example"""

from cyclops import Agent, AgentConfig
from cyclops.toolkit import tool
from datetime import datetime
import random


@tool
def get_time() -> str:
    """Get the current time"""
    return datetime.now().strftime("%H:%M:%S")


@tool
def get_weather(location: str) -> str:
    """Get weather for a location"""
    conditions = ["Sunny", "Cloudy", "Rainy", "Snowy"]
    temp = random.randint(50, 90)
    return f"Weather in {location}: {random.choice(conditions)}, {temp}°F"


@tool
def calculate(operation: str, a: float, b: float) -> str:
    """Perform basic math calculations"""
    ops = {
        "add": a + b,
        "subtract": a - b,
        "multiply": a * b,
        "divide": a / b if b != 0 else "Error: Division by zero",
    }
    result = ops.get(operation, "Unknown operation")
    return f"{a} {operation} {b} = {result}"


# Create agent with tools
config = AgentConfig(model="ollama/qwen3:4b")
agent = Agent(config, tools=[get_time, get_weather, calculate])

# Ask questions that require tools
print("Agent with tools demo:\n")

response = agent.run("What time is it?")
print(f"Q: What time is it?\nA: {response}\n")

response = agent.run("What's the weather in New York?")
print(f"Q: What's the weather in New York?\nA: {response}\n")

response = agent.run("What's 15 multiplied by 7?")
print(f"Q: What's 15 multiplied by 7?\nA: {response}\n")