Routine Cookbook

Routine Cookbook

Everything needed to take a task description → a working, tested routine. Read this overview, then fetch the companion file(s) for what your routine actually does:

manage_skill(action="read_file", name="routine_cookbook", file="hummingbot_client.md")

Which companion file to read

Most routines need report_builder.md plus one or two others. A continuous price monitor with a live dashboard, for example, reads hummingbot_client.md

• continuous.md.

First, ask what already exists

Before writing a line, find out what Condor already has. The index is generated from the code, so it cannot be out of date, and it costs nothing until you read it:

run_code(code="""
from condor.primitives import catalog, describe
print(catalog())                    # every fetcher + every routine, grouped
print(catalog("market_data"))       # one group only
print(describe("market_data.fetch_historical_candles"))   # signature + docstring
print(describe("routine:arb_check"))                      # config fields + defaults
""")

The same three imports work inside a routine. Never guess a signature — a wrong one costs a failed run, a traceback and a retry; describe() costs a line.

Compose instead of reimplementing

A routine can call another routine and use its result:

from condor.primitives import call_routine, start_routine
import asyncio

snap, pools = await asyncio.gather(
    call_routine("portfolio_snapshot"),
    call_routine("solana_pool_scanner", {"min_tvl": 250_000}),
)
print(snap.text, snap.report_id)

• call_routine(name, config) runs it inline and returns its RoutineResult (plus a report_id attribute for the report it saved). It is an implementation detail of your routine: no dock instance, no post-run hook, no message to the user. Chains are capped at depth 3 and cycles are refused.
• start_routine(name, config) runs it as a real background run and returns an instance_id — the run shows up in the dock, fires its hooks and reports back to the user. Use it for something slow, continuous, or that the user should see; read it back with manage_routines(action="get_instance", name=<instance_id>).
• Continuous routines can only be started, never called inline.

First: is this a routine at all?

A routine is a durable artifact — it has a name, a config schema, a place in the library, and it can be scheduled, shared and re-run by anyone. That is worth a file when the work repeats.

A one-off computation is not. "What were SOL's hourly returns yesterday", "what is the spread between these two venues right now", "aggregate these executors by controller" — write the Python and call run_code(code="..."). It runs in the bot with exactly the primitives below (context, client, pandas, pandas_ta, ReportBuilder, every condor.* module, and condor.primitives to find the rest), returns its print output and its result, and hands you the traceback to fix when it fails. No file, no Config class, no library entry.

Technical analysis: use pandas_ta, never hand-rolled math. Condor pins the same version the hummingbot-api image runs (pandas-ta>=0.4.71b → 0.4.71b0), so an RSI, EMA, ATR, MACD or Bollinger band computed in a routine is bit-for-bit the one a Hummingbot controller or a backtest computes. Writing your own EMA instead silently disagrees with the strategy you are analysing. Both call styles work — pandas_ta.rsi(df["close"], length=14) and the accessor df.ta.macd(append=True). In 0.4.x the multi-output column names repeat the deviation, e.g. BBP_20_2.0_2.0, MACD_12_26_9 — read the columns off the returned frame rather than assuming a 0.3.x name.

Promote a snippet to a routine when you have run essentially the same thing a third time, or the moment it needs to be scheduled, shared, or visible to the user.

Where the routine lives

Agent-local — agents/{slug}/routines/ — visible only to that agent, and shared across all of its strategies (there is no per-strategy library). This is the default when you are an agent: your own routines are yours, and you create them with no agent argument (your slug is already the scope).

Global — routines/ — visible to every user and agent. Use it only for general-purpose analysis/monitoring not tied to one agent. From the chat, target an agent's local dir explicitly with agent="<agent_slug>".

If the scope is ambiguous, clarify it before writing code.

Basic routine anatomy

from pydantic import BaseModel, Field
from telegram.ext import ContextTypes
from config_manager import get_client
import logging

logger = logging.getLogger(__name__)

CATEGORY = "Market Data"  # Market Data | Analysis | Arbitrage | Monitoring | Bot Analysis

class Config(BaseModel):
    """One-line description shown in UI."""
    trading_pair: str = Field(default="BTC-USDT", description="Trading pair")
    connector_name: str = Field(default="binance_perpetual", description="Exchange")

async def run(config: Config, context: ContextTypes.DEFAULT_TYPE) -> str:
    client = await get_client(context._chat_id, context=context)
    if not client:
        return "No server available"
    # ... work ...
    return "result string"

Must export: Config (Pydantic BaseModel) and async def run(config, context) -> str. The Config docstring is the UI description. CATEGORY groups it in the catalog.

The loop: create → test → fix

1. Understand — what to analyze, monitor or compute; agent-local or global?
2. Check existing — manage_routines(action="list") to avoid duplicates.
3. Read — this overview + the companion file(s) for what you are building.
4. Create — manage_routines(action="create_routine", name="snake_case", code="...")
5. Test — manage_routines(action="run", name="snake_case", config={})
6. Iterate — read the error, fix, re-run until the output is clean.

Never report a routine as done before step 5 comes back clean.

manage_routines action reference

manage_routines(action="list")
manage_routines(action="create_routine", name="x", code="...")
manage_routines(action="read_routine", name="x")
manage_routines(action="edit_routine", name="x", code="...")
manage_routines(action="delete_routine", name="x")
manage_routines(action="run", name="x", config={})       # one-shot
manage_routines(action="start", name="x", config={})     # continuous
manage_routines(action="stop", name="instance_id")       # stop continuous
manage_routines(action="list_instances")                 # list running

# From the chat, target an agent's local library by adding agent="<slug>"
manage_routines(action="create_routine", name="x", code="...", agent="agent_slug")
manage_routines(action="run", name="x", agent="agent_slug", config={})

Non-negotiables (apply to every routine)

• Every routine MUST generate a ReportBuilder report — see report_builder.md.
• NEVER wrap the report block in try/except. No except Exception: logger.warning("Report generation failed"). A swallowed report error makes the run look completed while no report exists — the failure must reach the runner. Same for the chart code that feeds the report.
• builder.source("routine", "<file name>") on every builder — that string is how the Routines page finds the report; wrong or missing and the report is saved but invisible. Bare file name, never agent_slug/name.
• All client calls are async — always await; never time.sleep, only asyncio.sleep.
• Parse defensively where you read external data: handle None/missing keys and return an error string. This covers API responses, not your own report code — never turn it into a blanket except over the routine body.
• Look it up, don't guess — describe("<ref>") before writing any call whose exact signature you are not certain of, and catalog() before writing a fetch you suspect already exists.
• One routine per task. Lead with code, be direct.
• Test after writing (manage_routines(action="run", ...)) and fix until the output is clean.

These are starting patterns, not a bypass — running a routine still goes through the normal execution/confirmation controls.