---
title: Your own tool loop
description: Shape C in full — a hand-rolled ReAct loop over a raw provider client, connected for experiments, monitoring and protection. Worked on a real production agent.
---

Most agents past a prototype are this shape: a loop you wrote, a tool registry
of your own, and a raw client — `openai`, `httpx`, `anthropic` — underneath.
Nothing patches it, because there is no constructor to wrap. So the three
things the SDK does for a framework become three things the doorway does:
**offer the attached tools, emit the spans, gate the calls.**

This page is the complete pattern, taken from a production patient-messaging
agent: a ReAct loop in `react.py`, its own `ToolRegistry`, Azure OpenAI over
raw `httpx`, inbound SMS through a webhook and a Celery worker. Every step
below was needed; none is decoration.

## What you touch, and what you do not

| | file | why |
| --- | --- | --- |
| **new** | `agents.py` | the doorway — experiments and the `@observe` handler |
| **one call** | wherever inbound messages are handled | it has to *call* `handle_message` |
| **two spans** | inside the loop | a hand-rolled loop emits nothing by itself |

The loop's control flow, its yields, its registry — untouched.

## 1. Identify the entry point

The function that runs one user turn. It takes the message and returns or
streams the answer. Read its signature; do not modify it.

```python
async def stream_react_response(
    session_id: str, agent_id: str, user_message: str,
    phone_number: str = "", system_directive: str = "", ...
) -> AsyncGenerator[dict, None]:
    # yields {"type": "chunk"|"thought"|"action"|"observation"|"final"|"error"|...}
```

Two things to note about it, because both decide the doorway's shape: it is
an **async generator** (so the doorway drains it), and it already accepts a
per-turn prompt addition (`system_directive`) — which is where the attached
tools get announced.

> **The doorway runs once per turn**
>
> In a conversation experiment Redline calls `run(task, ctx)` **once per
> message** — up to the experiment's turn limit — with the conversation so far
> in `task.prompt`, the way a stateless model API is called. Eighty turns is
> eighty calls. Build the agent once (the session cache below does that) and
> keep anything that *describes* the agent idempotent; nothing that grows
> belongs inside `run`.

## 2. The doorway — experiments

```python title="agents.py"
import os, uuid
from redline import agent, observe, redline_tools
from react_agent.tools.base import BaseTool, ToolResult          # your registry's interface
from routers.react_chat_router import get_or_create_session, stream_react_response


class _Attached(BaseTool):
    """An attached tool in the registry's own shape — advertised from
    as_openai_schema() and run through attached.call(), the two supported paths."""

    def __init__(self, attached, fn: dict):
        super().__init__(name=fn["name"], description=fn.get("description", ""))
        self._attached = attached
        self._parameters = fn.get("parameters") or {"type": "object", "properties": {}}

    def get_schema(self) -> dict:
        return self._parameters

    async def execute(self, parameters: dict, context=None) -> ToolResult:
        try:   # 5b — a failing tool is an observation, never a dead run
            return ToolResult(success=True, output=await self._attached.call(self.name, parameters or {}))
        except Exception as exc:
            return ToolResult(success=False, output="", error=f"{type(exc).__name__}: {exc}")


async def _run(task, ctx) -> str:
    attached = await redline_tools(task)                              # §4 Shape C
    try:
        agent_id = os.environ.get("CIRCA_AGENT_ID", "").strip()
        if not agent_id:
            raise RuntimeError("CIRCA_AGENT_ID is not set — which configured agent should experiments run against?")

        session_id = f"redline-{task.run_id or uuid.uuid4().hex}"
        phone = f"+1555{abs(hash(session_id)) % 10_000_000:07d}"    # one fresh agent per run
        live, *_ = await get_or_create_session(agent_id, session_id, phone, is_test=True)

        brief = ""
        if attached.list:
            for schema in attached.as_openai_schema():                # what to advertise
                live.agent.add_tool(_Attached(attached, schema["function"]))
            brief = "## Tools this experiment attached\n" + "\n".join(   # 5a
                f"- {t.name}: {t.description}" for t in attached.list)

        final, thought = "", []
        def flush():
            if thought:
                ctx.thinking("".join(thought).strip()); thought.clear()

        async for ev in stream_react_response(session_id=session_id, agent_id=agent_id,
                                              user_message=task.prompt, phone_number=phone,
                                              system_directive=brief, is_test=True):
            kind = ev.get("type")                                     # 5c — the stream onto ctx
            if kind in ("chunk", "thought"):
                thought.append(ev.get("content", "") or "")
                if kind == "thought": flush()
            elif kind == "action":
                flush(); ctx.tool(ev.get("tool", ""), ev.get("input"))
            elif kind == "observation":
                ctx.tool_result("", ev.get("content", ""))
            elif kind == "final":
                thought.clear(); final = ev.get("content", "") or ""
            elif kind == "error":
                flush(); ctx.log(f"error: {ev.get('message', '')}")
        flush()
        return final                                                  # §3 — the whole text, at the end
    finally:
        await attached.close()


@agent(id="circa-adaptive", name="CIRCA Adaptive Agent",
       description="Patient-facing ReAct agent over SMS and chat.")
async def run(task, ctx):
    return await _run(task, ctx)          # async entry point: just await it
```

What each part is for:

- **Attached tools go into your registry as native functions.** The model then
  picks them the same way it picks yours — no line protocol in the prompt.
  `as_openai_schema()` supplies the name, description and parameters;
  `attached.call()` executes by name. Wrapping those two in your registry's
  interface is the adapter; you are not re-implementing either.
- **The brief rides on the per-turn prompt hook you already have.** Registering
  a tool is not the same as the agent knowing to reach for it. Generated from
  the list, so it names whatever was attached.
- **One fresh agent per run.** Live agents here are keyed by caller, so a
  synthetic caller per run gives each run its own memory. `is_test=True` flags
  it the way the app's own test surfaces do.
- **Config the entry point needs is an environment variable with a clear
  error**, not a guess. Which configured agent to run against is the
  repository's knowledge, not the SDK's.

> **redline dev loads all of .env**
>
> Not only `REDLINE_*`. So `CIRCA_AGENT_ID` — and your model keys — can live
> in the same `.env` at the repository root, and the worker has them on start.

## 3. The doorway — production

```python title="agents.py"
@observe(agent="circa-adaptive", session_arg="session_id",
         user_arg="phone_number", input_arg="text")
async def handle_message(session_id: str, phone_number: str, text: str,
                         agent_id: str, state: dict | None = None) -> str:
    reply = ""
    async for ev in stream_react_response(session_id=session_id, agent_id=agent_id,
                                          user_message=text, phone_number=phone_number):
        kind = ev.get("type")
        if kind == "final":
            reply = ev.get("content", "") or ""
        elif kind == "session_ended" and state is not None:
            state["ended"] = True                 # what the caller read off the stream before
    if state is not None:
        state["text"] = reply
    return reply
```

Two rules made this shape:

**`@observe` wraps a coroutine that returns the answer — not a generator.** It
tests `inspect.iscoroutinefunction`, which is `False` for an async generator;
decorate the generator directly and it records the generator *object* as the
answer. So the handler drains the stream and returns the text.

**The existing pipeline must call it.** Here inbound SMS arrives at a webhook,
is queued to Celery, and a task drains the stream inline. That inline drain
becomes one call:

```python title="tasks/chat_tasks.py"
async def _drain_react():
    from agents import handle_message
    await handle_message(session_id=session_id, phone_number=user_phone_number,
                         text=user_msg_en, agent_id=str(agent_id), state=_react)
```

The `asyncio.run` around it, and everything that reads `_react["text"]` and
`_react["ended"]` afterwards, is unchanged. `state` exists precisely so the
caller keeps every signal it had before. **This is the one change to existing
code the integration asks for**, and without it production sessions never
appear — a decorator only sees calls to the function it is on.

## 4. The spans — what makes the waterfall

Under `redline dev`, or `@observe`, the SDK is listening for OpenTelemetry
spans. A hand-rolled loop emits none, and there is no instrumentation package
for a loop nobody else wrote. So the session shows every conversation turn and
nothing between — **"0 tool calls"**, no model calls, no tokens.

Two spans fix it, in the OTel GenAI convention the SDK already reads:

```python title="react_agent/otel.py"
from opentelemetry import trace
_TRACER = trace.get_tracer("circa.react_agent")

def turn_span():                               # one per user message — the parent
    return _TRACER.start_span("agent turn")

def llm_span(model, parent):                   # one per model call
    span = _TRACER.start_span("chat", context=trace.set_span_in_context(parent))
    span.set_attribute("gen_ai.operation.name", "chat")
    span.set_attribute("gen_ai.request.model", model)
    return span

def end_llm_span(span, prompt_tokens, completion_tokens):
    span.set_attribute("gen_ai.usage.input_tokens", prompt_tokens)
    span.set_attribute("gen_ai.usage.output_tokens", completion_tokens)
    span.end()

def tool_span(name, arguments, parent):        # one per tool call
    span = _TRACER.start_span(f"execute_tool {name}", context=trace.set_span_in_context(parent))
    span.set_attribute("gen_ai.operation.name", "execute_tool")
    span.set_attribute("gen_ai.tool.name", name)
    span.set_attribute("gen_ai.tool.call.arguments", json.dumps(arguments))
    return span

def end_tool_span(span, result):
    span.set_attribute("gen_ai.tool.call.result", str(result)[:8000])
    span.end()
```

And in the loop — the model call wraps the streaming read so the failover path
is covered too, and ends in a `finally` because a cycle that raises still
happened:

```python title="react_agent/react.py"
async def run_stream(self, user_input, system_directive=""):
    span = turn_span(); self._otel_turn = span
    try:
        async for event in self._run_stream(user_input, system_directive):
            yield event
    finally:
        self._otel_turn = None; span.end()

# inside the loop:
_span = llm_span(self.model, self._otel_turn)
try:
    async for chunk in self._stream_llm(messages, tools):
        ...                                      # accumulate usage from the final chunk
finally:
    end_llm_span(_span, usage["prompt_tokens"], usage["completion_tokens"])

_tspan = tool_span(tool_name, tool_input, self._otel_turn)
result = await self.registry.execute(tool_name, tool_input, self.context)
end_tool_span(_tspan, result.to_observation())
```

Three things that are not optional:

- **The per-turn parent.** A watcher reads the span tree to know where one
  turn ends. Flat, parentless spans are read as one turn each — eighteen
  spans from one message were drawn as eighteen turns until this was added.
- **`start_span()`, not `start_as_current_span()`.** The loop yields; a span
  made current across a `yield` leaks its context into whatever runs next.
  Parenting is explicit via `set_span_in_context`, which also nests correctly
  under `@observe`'s conversation span in production.
- **The turn span carries no `gen_ai.*` attribute.** Tag it as a model call
  and it renders as an empty LLM call; unclaimed, it is exactly what it is —
  the container.

Proof it lands, through the SDK's own adapter:

```
[log]  LLM call · gpt-4o · 439 tokens
[tool] lookup_patient_data {"phone": "+1555"}
[tool] lookup_patient_data → found 1 record
```

`opentelemetry-api` and `-sdk` are the only requirement, and `redline dev`
installs the span processor on the global tracer itself.

## 5. Protection — what applies and what does not

Runtime enforcement patches the `openai` and `litellm` clients. **A loop over
raw `httpx` uses neither, so the patch gates nothing here.** Policies, the
guard and honeypots still arm in the `redline dev` process and govern
experiment runs that go through a patched client — but a raw-HTTP production
loop is not covered by "nothing to write". If you need the gate there, the
SDK's `policy.gate_tool_call`, `guard.judge` and `honeypot.armed()` are the
calls; where they sit in your loop is a design choice we would rather talk
through than have you guess.

### 5a. The snapshot — declare what the agent is made of

Protect's first job is to know the agent: its system prompt and its tool
list, captured from inside the process, hashed, pinned, and scanned. Every
automatic capture reads a framework — a Pydantic AI or LangChain constructor,
an Agno or CrewAI object in your module, the `tools=[…]` a patched
openai/litellm client was handed. A loop of your own passes through none of
them, so **Protect shows "waiting for first snapshot" no matter how many runs
it has done**. Nothing is wrong; nothing was ever captured.

The hand-off is one call, once the agent is built, with the prompt and tools
it will actually run with:

```python title="agents.py"
from redline import declare_assets

async def _run(task, ctx) -> str:
    attached = await redline_tools(task)
    try:
        session = await get_or_create_session(...)
        built = session.agent
        # What the agent is made of: its own prompt and its own registry. The
        # attached tools are left out — this runs once per turn, and from turn
        # two the session already holds them; they also differ per experiment.
        # The pinned baseline is the agent, not the experiment.
        own = [t for t in built.registry.tools() if not isinstance(t, _Attached)]
        # The TEMPLATE, not the rendered turn — this builder bakes today's
        # date in, and a prompt that changes at midnight is a scan every day.
        prompt = built.build_system_prompt().replace(date.today().isoformat(), "{current_date}")
        declare_assets(system_prompt=prompt, tools=own)
        for schema in attached.as_openai_schema():
            built.add_tool(_Attached(attached, schema["function"]))
        ...
```

`tools` takes what a registry already produces: OpenAI function-tool dicts,
flat `{"name", "description", "parameters"}` dicts, or objects with `name`,
`description` and a `get_schema()` / `parameters`. It is best-effort and
silent, like every capture — a shape it cannot read is an asset missing from
the snapshot, never an exception in your run. The snapshot rides the next
keepalive (≤ 20 s); a changed prompt or tool list on a later run is a changed
hash, and the change monitor opens a review.

Four rules:

- **Declare the template, not the rendered turn.** A scan runs when the
  hash changes, and the hash is over exactly what you declare. A prompt with
  today's date in it is a new version at midnight; one with the patient's
  name in it is a new version per conversation — and each is a scan, with
  nothing real changed. Strip anything volatile (dates, names, session ids,
  fetched context) before declaring, as above. What Protect should see is
  what the agent *is*, not what it said this turn.
- **`run()` is called once per turn, so this runs once per turn.** A
  conversation experiment invokes your doorway once per message, with the
  conversation so far in `task.prompt` (see [§1](#1-identify-the-entry-point)).
  `declare_assets` is built for that: each argument you pass *replaces* the
  previous statement of that thing, so eighty turns produce one snapshot, not
  eighty. Do not use an SDK older than 0.2.27 for this — before it, the call
  accumulated, and one conversation produced two versions and two scans.
- **Declare the agent's own tools, not the attached ones.** Attached tools
  differ per experiment; declared with them, the baseline would change with
  every launch. Filter them out as above — on turn two the session already
  holds them.
- **Declare in the experiments doorway**, where `redline dev` is the
  process. The production handler runs in your worker, and nothing ships a
  snapshot from there.

While you are at it, name the framework. The SDK labels an agent by what is
importable beside it, and a venv that carries `langchain` for a retriever
labels a hand-rolled loop *langchain* — which reads as a promise the LangChain
patch will capture it. Say what it is:

```python
@agent(id="my-agent", name="My Agent", framework="custom")
```

## 6. Keep it running

The worker belongs in the same restart script as the app:

```bash title="restart.sh"
# stop
pkill -f 'redline dev' 2>/dev/null || true
# ... redis, celery, fastapi ...
# start — from the repo root, so it reads .env
nohup $PYTHON -m redline dev > $LOG_DIR/redline-dev.log 2>&1 &
disown
```

Non-fatal if it fails: the app runs, the agent shows offline.

## Checklist

- `agents.py` exports `run` (`@agent`) and `handle_message` (`@observe`)
- attached tools in the registry via `as_openai_schema()` + `call()`
- brief generated from `attached.list`, on the per-turn prompt hook
- entry-point config (`CIRCA_AGENT_ID`) in `.env`, with a clear error when absent
- the existing message pipeline calls `handle_message`
- one turn span, model spans and tool spans parented to it
- `declare_assets(system_prompt=…, tools=…)` in the doorway, before attached tools merge; `framework="custom"` on `@agent`
- `redline dev` in the restart script