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, 72F in {location}"

Both forms work, with or without parentheses:

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

To override the name or description, pass keyword arguments:

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

BaseTool subclass

For tools that need to hold state or require 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()

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

Sync vs async

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]

Type annotations and JSON schema

Cyclops converts Python type annotations to JSON schema types automatically.

Python type	JSON schema type
str	"string"
int	"integer"
float	"number"
bool	"boolean"
list	"array"
dict	"object"
unannotated	"string" (default)

Optional[T] is treated the same as T. The parameter becomes optional in the schema if it has a default value.

@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"))

This produces a JSON schema with operation as "string" and a, b as "number", all required.

Parameters with default values become optional in the schema:

@tool
def search(query: str, max_results: int = 5) -> str:
    """Search for information."""
    return f"Results for '{query}' (max {max_results})"

Here query is required and max_results is optional with a default of 5.

ToolRegistry

ToolRegistry is a named collection of tools. It is useful when managing 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 an existing tool manually:
registry.register(ping)

# Inspect:
print(registry.list_tools())       # ['ping']
print(registry.get_tool("ping"))   # <Tool ping>

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

Passing tools to Agent

Pass any list of tools (decorated functions or BaseTool instances) to the Agent constructor.

from cyclops import Agent, AgentConfig

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?"))

Pull tools from 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 (some Ollama models, for example), set tool_mode="naive" in AgentConfig. Cyclops also falls back to prompt-based tool invocation automatically when tool_mode="auto" and the API returns a function-calling error.

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")