Add a custom tool

Define a JSON schema, write an async handler, and register it with .on(). The return value reaches both the user and the model.

This guide is for developers connecting an agent to an API they own. The tool calls your service directly, so the base URL below is a placeholder for yours. You will add a tool that looks up an order, renders a result in the conversation, and shows the request and response in an expandable step.

Prerequisites: a working agent from Build your first agent.

1. Define the schema

A tool schema is a plain dict. Write the description for the model: say what the tool does and when to reach for it.

TOOLS = [
    {
        "name": "lookup_order",
        "description": (
            "Look up one order by its id. Use when the user asks about an order's "
            "status, contents or shipping. Returns the order and its line items."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "The order id, for example ORD-10293."},
            },
            "required": ["order_id"],
        },
    }
]

inputSchema and input_schema are both accepted.

2. Write the handler

A handler is an async function. Its return value is used twice: as the event streamed to the user, and as the tool_result sent back to the model.

import httpx

API = "https://api.your-company.com"   # your own service, not a Cycls endpoint


async def lookup_order(args):
    async with httpx.AsyncClient(timeout=20) as http:
        r = await http.get(f"{API}/orders/{args['order_id']}",
                           headers={"Authorization": f"Bearer {os.environ['ORDERS_KEY']}"})

    if r.status_code == 404:
        return f"No order named {args['order_id']}."
    if r.status_code >= 400:
        return f"Error: orders API returned {r.status_code}."

    order = r.json()
    lines = "\n".join(f"- {i['qty']} x {i['name']}" for i in order["items"])
    return f"**{order['id']}** ({order['status']})\n{lines}"

Returning a plain string is the simplest case. Returning a component dict renders that component instead:

async def render_chart(args):
    path = await build_chart(args["series"])
    return {"type": "image", "src": path, "caption": "Revenue by month"}

3. Register it

llm = (
    cycls.LLM()
    .model("anthropic/claude-sonnet-4-6")
    .system("You help customers with their orders.")
    .tools(TOOLS)
    .on("lookup_order", lookup_order, label=lambda i: i["order_id"])
)

Option	Effect
label	a function from input to string, shown on the step line. Defaults to the first string value in the input
icon	an image URL shown beside the step
details=True	the step expands into Request and Response, and the result goes there instead of into the conversation

.on("lookup_order", lookup_order, label=lambda i: i["order_id"], details=True)

Use details=True for tools whose raw output is long or uninteresting to read in the transcript. The model still receives the full result.

4. Use the caller’s identity

Declare a second parameter and the handler receives request context.

async def lookup_order(args, ctx):
    user = ctx.user               # the authenticated User
    ws = ctx.workspace            # the caller's workspace

    if not user:
        return "Sign in to look up orders."

    order = await orders.get(args["order_id"], customer=user.id)
    return format_order(order)

Attribute	Meaning
ctx.user	the authenticated user, or None
ctx.workspace	the active workspace scope
ctx.chat_id	the current chat
ctx.approvals	approvals granted this turn
ctx.auto	the composer’s automatic approval switch

One-argument handlers keep working unchanged.

5. Handle failure explicitly

The loop treats a returned string starting with Error: as a failed call, which is what the step indicator and the tool_call log record read.

async def lookup_order(args):
    try:
        ...
    except httpx.TimeoutException:
        return "Error: the orders API timed out. Tell the user to try again."

Write the message for the model, since the model is what reads it and decides what to do next.

Complete example

orders.py

import os

import cycls
import httpx

API = "https://api.your-company.com"   # your own service, not a Cycls endpoint

TOOLS = [
    {
        "name": "lookup_order",
        "description": (
            "Look up one order by its id. Use when the user asks about an order's "
            "status, contents or shipping."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"],
        },
    }
]


async def lookup_order(args, ctx):
    if not ctx.user:
        return "Sign in to look up orders."
    try:
        async with httpx.AsyncClient(timeout=20) as http:
            r = await http.get(
                f"{API}/orders/{args['order_id']}",
                headers={"Authorization": f"Bearer {os.environ['ORDERS_KEY']}"},
                params={"customer": ctx.user.id},
            )
    except httpx.TimeoutException:
        return "Error: the orders API timed out."

    if r.status_code == 404:
        return f"No order named {args['order_id']}."
    if r.status_code >= 400:
        return f"Error: orders API returned {r.status_code}."

    order = r.json()
    lines = "\n".join(f"- {i['qty']} x {i['name']}" for i in order["items"])
    return f"**{order['id']}** ({order['status']})\n{lines}"


llm = (
    cycls.LLM()
    .model("anthropic/claude-sonnet-4-6")
    .system("You help customers with their orders. Ask for an order id when you need one.")
    .tools(TOOLS)
    .on("lookup_order", lookup_order, label=lambda i: i["order_id"])
)


@cycls.agent(
    image=cycls.Image().pip("httpx").copy(".providers.env", ".env"),
    web=cycls.Web().auth(cycls.Clerk()).title("Order support"),
    volumes={"/workspace": cycls.Volume("orders-chats")},
)
async def orders(context):
    async for ev in llm.run(context=context):
        yield ev

When to use a connector instead

Write a custom tool when your code holds the credential, such as a service key that belongs to the deployment. Use a connector when the credential belongs to the end user, such as their own Notion or GitHub account. Connectors handle the grant, refresh, per-tool approvals and the audit trail.

Next

Serve an app

A FastAPI service with sign-in and per-user storage.