Skip to content
Redline
Esc
↑↓navigate↵open⌘Jpreview
On this page

Agno

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:

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

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": "<tool>", "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()
  • 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:

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:

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.

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:

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.

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 <tool> was denied by policy: <reason>; 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.

Was this page helpful?