--- title: Agno description: Connect an Agno agent to Redline — experiments, production monitoring and runtime protection. The complete document, written to be followed by a coding agent inside the repository. --- You are connecting a repository whose agent is built on **Agno** to the Redline platform: experiments AND production monitoring. Follow these steps exactly and do not modify the repository's existing agent — the whole integration is ONE new file plus ONE instrumentation package. ## 0. Confirm the shape Grep for `agno`, `from agno.agent import Agent`. If present, this is the page. Agno is **Shape C**: NOT patched by the SDK, so the doorway merges the experiment's attached tools itself (§4). What sets Agno apart is the **event loop** — its db pools and knowledge stores stay open between calls, and the wrong doorway shape tears them down (§3). ## 1. Install and scaffold At the repository root, inside its virtualenv: ```bash pip install redlineai-sdk openinference-instrumentation-agno redline init ``` Install the instrumentation now, before the first `redline dev`. `redline init` writes `agents.py` — never `redline.py`. If the agent's own example depends on a database (pgvector, for instance), start it the way the repository already does before `redline dev`. ## 2. Find the agent's entry point Locate where the `Agent` is constructed and run: `agent.run(`, `agent.arun(`, `agent.print_response(`, or a factory like `create_support_agent(...)`. Read what it takes and whether it is async. **Do NOT modify it.** ## 3. Wire it into `agents.py` ```python title="agents.py" import json from redline import agent, redline_tools import my_app.support_agent as subject # unchanged async def _run(task, ctx) -> str: attached = await redline_tools(task) # §4 Shape C built = subject.create_support_agent(CUSTOMER, TICKET, ORG) prompt = task.prompt if attached.list: # 5a. Named from the list, so it describes whatever was attached. brief = "## Tools this experiment attached\n" + "\n".join( f"- {t.name}: {t.description}" for t in attached.list) prompt = f"{prompt}\n\n{brief}\n" ctx.log(f"attached {len(attached.list)} tools: {', '.join(attached.names())}") advertised = attached.as_openai_schema() # what to advertise if advertised: prompt += ('\nTo use one of the attached tools, reply with exactly one line:\n' 'REDLINE_TOOL {"name": "", "arguments": {…}}\n' "You will be given its output and may then continue.\n" f"Their schemas: {json.dumps(advertised)}\n") reply = getattr(await built.arun(prompt), "content", "") or "" for _ in range(12): # 5b — call() never raises line = next((l for l in reply.splitlines() if l.strip().startswith("REDLINE_TOOL")), None) if not line: break try: request = json.loads(line.split("REDLINE_TOOL", 1)[1].strip()) name, arguments = request.get("name", ""), request.get("arguments", {}) except Exception as exc: ctx.log(f"unparseable tool request: {exc}"); break ctx.tool(name, arguments) output = await attached.call(name, arguments) # what to run when picked ctx.tool_result(name, output) reply = getattr(await built.arun( f"You asked for {name}. Its output:\n{output}\n\nContinue."), "content", "") or "" return reply @agent(id="learning-support", name="Learning Support Agent", description="A support agent with user profile, session context and org-shared memory.") async def run(task, ctx): return await _run(task, ctx) # async straight through — NEVER asyncio.run() ``` > **async def run is not a style choice** > > `asyncio.run()` builds an event loop, runs one coroutine, and **destroys the > loop**. Agno keeps async resources — connection pools, knowledge stores — > alive between calls. On the second run they belong to a loop that no longer > exists, and the worker crashes with a closed-loop error that names nothing > useful. An `async def run` is executed by the SDK on a persistent loop it > owns, and the pools survive. - `id` is a kebab-case slug. `task.prompt` is the instructions. RETURN the final answer as a string. If `arun` streams, consume it and return the whole text. ## 4. Attached tools — you MUST merge them yourself An experiment can attach MCP servers and, on every run, the project's Linux machine as a `machine_run` shell tool. Agno has no constructor the SDK wraps, so: ```python attached = await redline_tools(task) attached.as_openai_schema() # what to advertise await attached.call(name, arguments) # what to run when picked ``` **Do not hand-write an adapter over `attached.list`**; these two are the supported path. If your Agno agent's `tools=[...]` can take plain callables, you may instead wrap each attached tool as a function that calls `attached.call(name, kwargs)` and pass those in at construction — that gives native function calling instead of the `REDLINE_TOOL` line. Attached SKILLS arrive inside `task.prompt`. ## 5. Three failures that are nearly guaranteed if you skip them **5a. Tell the agent its tools exist.** The brief in §3, GENERATED FROM THE LIST. Without it the agent answers "I have no way to do that". **5b. A tool that raises must not kill the run.** `attached.call()` returns errors as text. Wrap the agent's own tools the same way. **5c. Do not drop the reasoning.** `ctx.tool` / `ctx.tool_result` around each attached call; `ctx.log` for anything else. Agno's own model and tool calls arrive via the instrumentation (§8a). ## 6. Connect and verify The agent's environment must load the way the repo normally loads it. Then, in `.env` at the repo root or exported: ```bash REDLINE_URL=https://tryredlineai.co REDLINE_API_KEY=rl_… # Agents page → Runner key redline dev ``` `redline dev` reads `.env` itself — every variable in it. `redline dev` **must be running on this machine at all times**: experiments execute inside it, the agent shows ONLINE only while it runs, snapshots for Protect upload through it, policies and honeypots re-arm through it every 20 seconds, dev sessions stream to Monitor through it. - `REDLINE_URL` — the Redline platform's address; nothing here hosts it. - `REDLINE_API_KEY` — the runner key from the Agents page. **Do not commit it.** On a server or VM: `nohup redline dev > ~/redline-dev.log 2>&1 &` then `disown`. Outbound only. Success: `connected — learning-support registered and online`. ## 7. When something is wrong - **"no agents found"** → `agents.py` exports no `@agent` function. - **Registration rejected** → id collides; pick another slug. - **Worker crashes on the SECOND run, "Event loop is closed"** → `asyncio.run()` in the doorway. Use `async def run` (§3). - **The agent says it has no tools** → §4 not done, or §5a missing. - **The run dies on the first tool error** → §5b. - **Session shows only User/Agent rows, no model calls** → §8a package missing or installed after `redline dev` started. Install and restart. - **Database connection refused at start** → the agent's own store (pgvector) is not running; start it as the repository does. - **`ask_user` ENDS the run** — return the questions as the answer. Everything else: [known issues](/integrations/known-issues). ## 8. Monitoring — the same agent, watched in production Nothing extra to write for dev: while `redline dev` is up, every conversation ALSO streams to **Monitor → Sessions**. **8a. REQUIRED — `pip install openinference-instrumentation-agno`, before `redline dev` starts.** The SDK only LISTENS; Agno emits nothing without it. VERIFY after the first run: model calls and a token count on the session. **For the DEPLOYED agent** — real users, no `redline dev`: ```python title="agents.py" from redline import observe @observe(agent="learning-support", session_arg="conversation_id", user_arg="user_id", input_arg="text") async def handle_message(conversation_id: str, user_id: str, text: str) -> str: built = subject.create_support_agent(CUSTOMER, conversation_id, ORG) return getattr(await built.arun(text), "content", "") or "" ``` Using `conversation_id` as the agent's session id keeps Agno's own memory and the Redline session aligned. Config is environment only; without `REDLINE_API_KEY` the decorator is inert. **8b. Your application has to CALL `handle_message`.** One call-site change in the existing message pipeline. Without it, production sessions never appear. **8c. It must return, not yield.** `@observe` does not wrap async generators. What the platform does with a session: Monitor → Sessions with the waterfall; violations per span and re-judged within minutes; intents mined on close. See [monitoring](/integrations/monitoring). ## 9. Runtime protection — what to expect once connected (nothing to write) Agno speaks to the model through the OpenAI client, which the SDK patches. Policies, the guard classifier and honeypots configured in the console reach the running process within ~20 seconds. A denied tool call is stripped from the model's response before Agno executes it and replaced with `[redline] The call to was denied by policy: `; the guard scores every user message and tool result; honeypots are injected beside the agent's own tools. Streaming passes through unjudged; everything fails OPEN.