Your own tool loop

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.

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.

2. The doorway — experiments

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.

3. The doorway — production

@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:

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:

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:

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:

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). 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:

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

6. Keep it running

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

# 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