Configuring an agent in ravnar¶

This tutorial explains how to configure an agent in ravnar. You will see two approaches:

1. Full control — subclassing the Agent ABC directly.
2. Using a wrapper — adapting an existing pydantic-ai agent via PydanticAiAgentWrapper, and giving it the tools of an MCP server.

pydantic-ai is used throughout as the example, but ravnar is not tied to it: under the hood ravnar is an AG-UI server, so any agent that emits AG-UI events is a first-class citizen. Other built-in options are summarised at the end.

A special Client is used for the documentation. For real-world scenarios, it can be substituted with a regular HTTP client with the base URL set to the URL of your ravnar deployment.

In [1]:

import json
import uuid
from collections.abc import AsyncIterator

from _ravnar.docs import Client


def print_json(obj):
    print(json.dumps(obj, indent=2, sort_keys=False))


def run_agent(client, agent_id: str, message: str) -> None:
    """Send a message to an agent and display the response in a human-readable format."""
    import httpx_sse

    body = {
        "thread_id": str(uuid.uuid4()),
        "run_id": str(uuid.uuid4()),
        "state": None,
        "tools": [],
        "context": [],
        "forwardedProps": None,
        "messages": [
            {
                "role": "user",
                "content": [{"type": "text", "text": message}],
                "id": str(uuid.uuid4()),
            }
        ],
    }

    with httpx_sse.connect_sse(
        client, "POST", f"/api/agents/{agent_id}/run", json=body
    ) as event_source:
        text = ""
        for sse in event_source.iter_sse():
            event = json.loads(sse.data)
            match event["type"]:
                case "TEXT_MESSAGE_CONTENT":
                    text += event["delta"]
                case "TOOL_CALL_START":
                    print(f"  🛠  Calling tool: {event['toolCallName']}")
                case "TOOL_CALL_RESULT":
                    print(f"  ✅ {event['content']}")
                case "RUN_ERROR":
                    print(f"  ❌ Error: {event.get('error', 'Unknown error')}")

        if text:
            print(f"\n{text}")

import json import uuid from collections.abc import AsyncIterator from _ravnar.docs import Client def print_json(obj): print(json.dumps(obj, indent=2, sort_keys=False)) def run_agent(client, agent_id: str, message: str) -> None: """Send a message to an agent and display the response in a human-readable format.""" import httpx_sse body = { "thread_id": str(uuid.uuid4()), "run_id": str(uuid.uuid4()), "state": None, "tools": [], "context": [], "forwardedProps": None, "messages": [ { "role": "user", "content": [{"type": "text", "text": message}], "id": str(uuid.uuid4()), } ], } with httpx_sse.connect_sse( client, "POST", f"/api/agents/{agent_id}/run", json=body ) as event_source: text = "" for sse in event_source.iter_sse(): event = json.loads(sse.data) match event["type"]: case "TEXT_MESSAGE_CONTENT": text += event["delta"] case "TOOL_CALL_START": print(f" 🛠 Calling tool: {event['toolCallName']}") case "TOOL_CALL_RESULT": print(f" ✅ {event['content']}") case "RUN_ERROR": print(f" ❌ Error: {event.get('error', 'Unknown error')}") if text: print(f"\n{text}")

Full control with the Agent ABC¶

The Agent abstract base class gives you complete control over the agent's behaviour. You implement a single method, run(), which receives the incoming RunAgentInput and a User object, and yields Events.

Let's build a simple agent that greets the current user.

In [2]:

import ag_ui.core

from ravnar.agents import Agent
from ravnar.authenticators import User


class WhoAmIAgent(Agent):
    """A simple agent that greets the current user."""

    async def run(
        self, input: ag_ui.core.RunAgentInput, user: User
    ) -> AsyncIterator[ag_ui.core.Event]:
        message_id = str(uuid.uuid4())

        yield ag_ui.core.RunStartedEvent(
            thread_id=input.thread_id,
            run_id=input.run_id,
            parent_run_id=input.parent_run_id,
        )
        yield ag_ui.core.TextMessageStartEvent(message_id=message_id)

        text = f"Hello, {user.id}!"
        for word in text.split():
            yield ag_ui.core.TextMessageContentEvent(
                message_id=message_id, delta=word + " "
            )

        yield ag_ui.core.TextMessageEndEvent(message_id=message_id)
        yield ag_ui.core.RunFinishedEvent(
            thread_id=input.thread_id, run_id=input.run_id
        )

import ag_ui.core from ravnar.agents import Agent from ravnar.authenticators import User class WhoAmIAgent(Agent): """A simple agent that greets the current user.""" async def run( self, input: ag_ui.core.RunAgentInput, user: User ) -> AsyncIterator[ag_ui.core.Event]: message_id = str(uuid.uuid4()) yield ag_ui.core.RunStartedEvent( thread_id=input.thread_id, run_id=input.run_id, parent_run_id=input.parent_run_id, ) yield ag_ui.core.TextMessageStartEvent(message_id=message_id) text = f"Hello, {user.id}!" for word in text.split(): yield ag_ui.core.TextMessageContentEvent( message_id=message_id, delta=word + " " ) yield ag_ui.core.TextMessageEndEvent(message_id=message_id) yield ag_ui.core.RunFinishedEvent( thread_id=input.thread_id, run_id=input.run_id )

The User object carries the authenticated user's identity along with any additional data and permissions. When no authenticator is configured, the user defaults to the current system user, and all permissions are granted.

Now we register it as a static agent through the ravnar configuration. Static agents are declared upfront and are available for the entire lifetime of the server.

In [3]:

config = {
    "agents": {
        "static": {
            "whoami": WhoAmIAgent,
        }
    }
}
client = Client(config)

config = { "agents": { "static": { "whoami": WhoAmIAgent, } } } client = Client(config)

Let's verify that our agent is registered and inspect its capabilities.

In [4]:

agents = client.get("/api/agents").raise_for_status().json()
print_json(agents)

agents = client.get("/api/agents").raise_for_status().json() print_json(agents)

[
  {
    "id": "whoami",
    "capabilities": {
      "transport": {
        "streaming": true
      }
    },
    "quickPrompts": []
  }
]

The capabilities are derived from the default get_capabilities() method of the base class which, among others, reports that the agent supports streaming. No tools are declared — this is a purely conversational agent.

Time to send it a message.

In [5]:

run_agent(client, "whoami", "Who am I?")

run_agent(client, "whoami", "Who am I?")

Hello, docs! 

The run_agent() helper parsed the SSE event stream and printed only the text content. The agent reads the user ID from the User object and includes it in the greeting.

Subclassing Agent is the most flexible approach — you have full control over the event stream and can integrate virtually any protocol or library. However, it also means you are responsible for producing the right events at the right time.

Using the Pydantic AI wrapper¶

If you already use pydantic-ai, you do not need to implement the Agent interface yourself. ravnar ships with PydanticAiAgentWrapper, which adapts any pydantic_ai.Agent into a ravnar agent — handling event generation, tool call streaming, and capability detection automatically.

We'll give a single agent two kinds of tool: an in-process Python function, and the tools of an MCP server.

First, the in-process tool. whoami is a regular async function that takes RunContext[User] as its first parameter — ravnar injects the authenticated User as the dependency when the agent runs.

In [6]:

from pydantic_ai import RunContext


async def whoami(ctx: RunContext[User]) -> str:
    """Get the current user's identity."""
    return ctx.deps.id

from pydantic_ai import RunContext async def whoami(ctx: RunContext[User]) -> str: """Get the current user's identity.""" return ctx.deps.id

Second, an MCP server. pydantic-ai connects to one with MCPToolset, and because the wrapper introspects the agent's toolsets, the server's tools are discovered and called exactly like whoami — no extra wiring on the ravnar side.

One difference matters: an MCP server runs as a separate process (or a remote service), so its tools don't automatically receive ravnar's injected User the way an in-process tool does — though you can forward it explicitly if needed. Reach for an in-process tool when a tool needs the caller's identity, and an MCP server when the capability is self-contained.

Let's write a minimal stdio MCP server that exposes a single add tool. In a real project this would be a separate service; here we write it to a temporary file so the tutorial stays self-contained.

In [7]:

import pathlib
import tempfile

mcp_server = pathlib.Path(tempfile.mkdtemp()) / "calculator.py"
mcp_server.write_text(
    '''
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("calculator")


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b


if __name__ == "__main__":
    mcp.run()
'''
)

import pathlib import tempfile mcp_server = pathlib.Path(tempfile.mkdtemp()) / "calculator.py" mcp_server.write_text( ''' from mcp.server.fastmcp import FastMCP mcp = FastMCP("calculator") @mcp.tool() def add(a: int, b: int) -> int: """Add two numbers.""" return a + b if __name__ == "__main__": mcp.run() ''' )

Out[7]:

202

Now we register the agent purely through configuration — no Python instantiation needed. ravnar's ImportStringWithParams mechanism resolves nested definitions recursively, so we declare the whole agent tree — model, wrapper, the in-process tools, and the MCP toolsets — as a single config block. MCPToolset takes the path to a script and launches it as a stdio server (it also accepts a URL for a remote HTTP/SSE server).

In [8]:

config = {
    "agents": {
        "static": {
            "assistant": {
                "cls_or_fn": "ravnar.agents.PydanticAiAgentWrapper",
                "params": {
                    "agent": {
                        "cls_or_fn": "pydantic_ai.Agent",
                        "params": {
                            "model": {
                                "cls_or_fn": "pydantic_ai.models.test.TestModel",
                                "params": {"call_tools": "all"},
                            },
                            "deps_type": User,
                            "tools": [whoami],
                            "toolsets": [
                                {
                                    "cls_or_fn": "pydantic_ai.mcp.MCPToolset",
                                    "params": {"client": str(mcp_server)},
                                }
                            ],
                        },
                    },
                },
            },
        },
    },
}
client = Client(config)

config = { "agents": { "static": { "assistant": { "cls_or_fn": "ravnar.agents.PydanticAiAgentWrapper", "params": { "agent": { "cls_or_fn": "pydantic_ai.Agent", "params": { "model": { "cls_or_fn": "pydantic_ai.models.test.TestModel", "params": {"call_tools": "all"}, }, "deps_type": User, "tools": [whoami], "toolsets": [ { "cls_or_fn": "pydantic_ai.mcp.MCPToolset", "params": {"client": str(mcp_server)}, } ], }, }, }, }, }, }, } client = Client(config)

The wrapper builds the capability object dynamically by introspecting the underlying pydantic-ai agent (via extract_capabilities()), so both tools show up: the argument-less whoami, and add with a parameter schema discovered from the MCP server.

In [9]:

agents = client.get("/api/agents").raise_for_status().json()
print_json(agents)

agents = client.get("/api/agents").raise_for_status().json() print_json(agents)

[
  {
    "id": "assistant",
    "capabilities": {
      "identity": {
        "type": "pydantic-ai"
      },
      "transport": {
        "streaming": true
      },
      "tools": {
        "supported": true,
        "items": [
          {
            "name": "whoami",
            "description": "Get the current user's identity.",
            "parameters": {
              "additionalProperties": false,
              "properties": {},
              "type": "object"
            }
          },
          {
            "name": "add",
            "description": "Add two numbers.",
            "parameters": {
              "properties": {
                "a": {
                  "title": "A",
                  "type": "integer"
                },
                "b": {
                  "title": "B",
                  "type": "integer"
                }
              },
              "required": [
                "a",
                "b"
              ],
              "title": "addArguments",
              "type": "object"
            }
          }
        ],
        "clientProvided": true
      },
      "output": {
        "structuredOutput": false
      }
    },
    "quickPrompts": []
  }
]

Let's run it. TestModel is pydantic-ai's stand-in for a real model: it exercises the full agent plumbing without calling an LLM, and with call_tools="all" it invokes every available tool once — both the in-process whoami and the MCP add.

In [10]:

run_agent(client, "assistant", "What is 2 + 3, and who am I?")

run_agent(client, "assistant", "What is 2 + 3, and who am I?")

  🛠  Calling tool: whoami
  🛠  Calling tool: add
  ✅ docs
  ✅ 0

{"whoami":"docs","add":0}

TestModel calls each tool with placeholder arguments, so add returns 0 rather than 5, and whoami returns your system username (no authenticator is configured). A real model would read the message and call add(a=2, b=3). The point is the plumbing: ravnar exposed an in-process tool and an MCP server's tool through the same wrapper, advertised both, and routed the calls — all from configuration. In production you would swap TestModel for a real model (e.g. openai, anthropic, openrouter); everything else stays the same.

To use a server you do not run yourself (the common case), pass its URL instead of a script path, e.g. {"cls_or_fn": "pydantic_ai.mcp.MCPToolset", "params": {"client": "https://example.com/mcp"}}.

Summary¶

• Subclass Agent directly when you need full control over the event stream or want to integrate a custom protocol.
• Use PydanticAiAgentWrapper when you already have a pydantic-ai agent — ravnar plugs it in automatically.
• ravnar injects the User object into the agent's run() method. For pydantic-ai agents, it is available as deps in tools via RunContext.deps.
• Add MCPToolset to a pydantic-ai agent's toolsets to expose the tools of any MCP server; the wrapper discovers and streams them automatically.
• Not using pydantic-ai? ravnar also ships AgnoAgentWrapper for Agno agents and SSEAgent to connect any agent that already speaks AG-UI over HTTP — and you can always implement the Agent ABC directly. See the Python API reference for the full list of built-in agents.
• All agents are registered through the same configuration mechanism, whether they are custom subclasses or wrappers.