Agent RPC & Actions

The Agent class includes a powerful RPC (Remote Procedure Call) system that allows you to expose methods as callable Actions. This brings a structured, ergonomic, and type-safe approach to agent communication, moving beyond simple message passing.

What is an Action?

An Action is a Python method on your Agent subclass that is decorated with @action. This decorator registers the method with the agent’s RPC system, making it callable by external clients (HTTP, WebSocket) and other agents.

from ai_query.agents import Agent, action

class TicketAgent(Agent):
    @action
    async def resolve(self, ticket_id: int, reason: str):
        # Your logic here...
        await self.update_state(...)
        return {"status": "resolved", "ticket_id": ticket_id}

How to Call Actions

Actions can be called through multiple transports, making them highly versatile.

1. Via REST API (HTTP)

AgentServer and AgentRouter automatically create a REST endpoint for each action.

URL: POST /agent/{agent_id}/action/{action_name}
Body: JSON payload with action parameters.

curl -X POST http://localhost:8080/agent/support/action/resolve \
     -H "Content-Type: application/json" \
     -d '{"ticket_id": 123, "reason": "Fixed."}'

2. Via WebSocket

For stateful clients, actions can be called over an existing WebSocket connection.

const callId = "req-1";
socket.send(JSON.stringify({
  type: "action",
  name: "resolve",
  params: { ticket_id: 123, reason: "Fixed." },
  call_id: callId
}));

The agent will send an action_result event with the same call_id upon completion.

3. Agent-to-Agent (Location Transparent RPC)

Use self.call() to get a type-safe proxy for communicating with another agent, regardless of whether it’s running locally or on a remote server. Local Example:

class ManagerAgent(Agent):
    async def escalate_ticket(self, ticket_id):
        # 1. Get proxy for the target agent
        # (Pass the class for type safety and autocompletion)
        support = self.call(
            "support-agent-1",
            agent_cls=SupportAgent,
            timeout=60,
            signal=self.context.signal,
        )

        # 2. Call the action fluently
        result = await support.resolve(
            ticket_id=ticket_id,
            reason="Escalated by manager."
        )
        return result

Remote Example: Now, imagine support-agent-1 is deployed as a serverless function. With the Agent Registry, your ManagerAgent code does not change.

# infrastructure.py
from ai_query import AgentRegistry, AgentServer, HTTPTransport

registry = AgentRegistry()

# Manager is local
registry.register("manager", ManagerAgent)

# Support is remote
registry.register(
    "support-agent-1",
    HTTPTransport("https://api.myapp.com/agents/support")
)

# Start the server
AgentServer(registry).serve()

The self.call("support-agent-1") inside ManagerAgent is automatically routed over HTTP, but your business logic remains clean and unaware of the network. This is Location Transparency. Local calls are routed through the target agent’s mailbox. The server creates and starts the target agent when needed, and the target handles one mailbox operation at a time. When a call supplies an AbortSignal, the same signal is available to the target action as self.context.signal, so delegated work can pass it to model turns or other cancellable operations. Set timeout=None for work whose lifecycle is bounded by the abort signal rather than a wall-clock deadline. After that signal aborts, local calls allow up to one second for cooperative target cleanup before ending the caller-side wait. Once the signal aborts, a target success returned during that cleanup grace period is discarded rather than overriding the abort.

Observe Local Call Events

Local calls can observe the target agent’s emitted events while its action is running. The handler receives each Event in emission order, and each handler call finishes before the delegated action can complete:

from ai_query import Event

async def forward_progress(event: Event) -> None:
    await progress_sink.send(event)

result = await self.call(
    "support-agent-1",
    agent_cls=SupportAgent,
    timeout=None,
    signal=self.context.signal,
    on_event=forward_progress,
).resolve(ticket_id=ticket_id, reason="Escalated by manager.")

The observer is scoped to that mailbox invocation and is deactivated on success, failure, timeout, or cancellation. Detached tasks retain their originating invocation context and cannot leak events into a later call. Observer exceptions are logged and do not replace the action’s primary result. Event handlers must not re-enter emit() on the target agent; attempted re-entry is rejected instead of deadlocking. Transports opt into this API with supports_call_events = True. HTTPTransport, DurableObjectTransport, and legacy custom transports do not support on_event; they raise NotImplementedError when it is requested.

Multi-User Metadata (Connection State)

A key feature of the RPC system is its awareness of the caller. You can store metadata on a per-connection basis without polluting the agent’s main state.

How it Works

on_connect: When a client connects, you can store information in connection.state.
self.context: When an action is executed, self.context.connection provides access to the state of the specific client who made the call.

Example: A Multi-User Chat Room

from ai_query.agents import Agent, action, ConnectionContext

class ChatRoom(Agent):
    async def on_connect(self, connection: Connection, ctx: ConnectionContext):
        """Store user_id for the connection."""
        user_id = ctx.metadata.get("user_id", "anonymous")
        connection.state["user_id"] = user_id
        await self.emit("status", {"text": f"{user_id} has joined."})

    @action
    async def who_am_i(self) -> str:
        """Returns the user_id of the caller."""
        # Access the user_id of the specific client calling this action
        user_id = self.context.connection.state.get("user_id", "unknown")
        return f"You are {user_id}."

In this example, if two different users call the who_am_i action, they will each get their own user_id back, because the agent can distinguish between them using self.context.

Start Here

Foundations

Agents

Deployment

Advanced

Agent RPC & Actions

What is an Action?

How to Call Actions

1. Via REST API (HTTP)

2. Via WebSocket

3. Agent-to-Agent (Location Transparent RPC)

Observe Local Call Events

Multi-User Metadata (Connection State)

How it Works

Example: A Multi-User Chat Room

​What is an Action?

​How to Call Actions

​1. Via REST API (HTTP)

​2. Via WebSocket

​3. Agent-to-Agent (Location Transparent RPC)

​Observe Local Call Events

​Multi-User Metadata (Connection State)

​How it Works

​Example: A Multi-User Chat Room

What is an Action?

How to Call Actions

1. Via REST API (HTTP)

2. Via WebSocket

3. Agent-to-Agent (Location Transparent RPC)

Observe Local Call Events

Multi-User Metadata (Connection State)

How it Works

Example: A Multi-User Chat Room