Agent - AgentScope

AgentScope provides several built-in agent types to cover different use cases. Use the cards below to jump to the section you need.

ReAct Agent

The primary agent for tool-using, reasoning, and structured output tasks.

Customizing Agents

Build your own agent by extending AgentBase or ReActAgentBase.

Agent Hooks

Inject custom logic before or after agent core functions.

State and Session Management

Save and restore agent state across sessions.

A2A Agent

Connect to remote agents using the Agent-to-Agent protocol.

Realtime Agent

Handle voice and live interactions with realtime model APIs.

ReAct Agent

ReActAgent is the primary built-in agent in AgentScope. It supports:

Feature	Description
Realtime steering	Interrupt the agent at any time during execution
Memory compression	Automatically compress long conversation history
Parallel tool calls	Execute multiple tool calls concurrently
Structured output	Return typed, schema-validated responses
Fine-grained MCP control	See MCP integration
Meta tool	Agent-controlled tool management
Long-term memory	Self-controlled persistent memory
Plan	Break complex tasks into managed subtasks and execute them systematically
Automatic state management	See State and Session Management

Realtime Steering

Realtime steering lets you interrupt an agent mid-reply. Call agent.interrupt() to cancel the current task. The agent then runs handle_interrupt() for post-processing.

import asyncio
from agentscope.agent import ReActAgent

async def main():
    # ... agent setup ...

    # Start the agent reply as a task
    reply_task = asyncio.create_task(agent(msg))

    # Interrupt after 1 second
    await asyncio.sleep(1)
    await agent.interrupt()

    result = await reply_task
    print(result.get_text_content())

You can override handle_interrupt in a subclass to customize the response when an interruption occurs — for example, calling the LLM to generate a context-aware acknowledgment.

Memory Compression

As conversations grow, token counts can exceed model limits. Enable automatic compression by passing a CompressionConfig when creating the agent:

from agentscope.agent import ReActAgent
from agentscope.token import CharTokenCounter

agent = ReActAgent(
    name="Assistant",
    sys_prompt="You are a helpful assistant.",
    model=model,
    formatter=formatter,
    compression_config=ReActAgent.CompressionConfig(
        enable=True,
        agent_token_counter=CharTokenCounter(),
        trigger_threshold=10000,  # compress when exceeding 10,000 tokens
        keep_recent=3,            # keep the 3 most recent messages uncompressed
    ),
)

When the token count exceeds trigger_threshold, the agent compresses older messages into a structured summary with these default fields:

Field	Description
`task_overview`	The user’s core request and success criteria
`current_state`	What has been completed, including files and outputs
`important_discoveries`	Constraints, decisions, errors, and failed approaches
`next_steps`	Specific actions needed to complete the task
`context_to_preserve`	User preferences, domain details, and promises made

Compression uses a marking mechanism — old messages are marked as compressed and excluded from future retrievals, while the summary is stored separately. Original messages are preserved.

Customizing compression You can control the compression behavior with summary_schema, summary_template, and compression_prompt:

from pydantic import BaseModel, Field

class CustomSummary(BaseModel):
    main_topic: str = Field(max_length=200, description="The main topic of the conversation")
    key_points: str = Field(max_length=400, description="Important points discussed")
    pending_tasks: str = Field(max_length=200, description="Tasks that remain to be done")

agent = ReActAgent(
    name="Assistant",
    sys_prompt="You are a helpful assistant.",
    model=model,
    formatter=formatter,
    compression_config=ReActAgent.CompressionConfig(
        enable=True,
        agent_token_counter=CharTokenCounter(),
        trigger_threshold=10000,
        keep_recent=3,
        summary_schema=CustomSummary,
        compression_prompt=(
            "<system-hint>Summarize the conversation focusing on "
            "the main topic, key points, and pending tasks.</system-hint>"
        ),
        summary_template=(
            "<system-info>Summary:\n"
            "Main Topic: {main_topic}\n\n"
            "Key Points:\n{key_points}\n\n"
            "Pending Tasks:\n{pending_tasks}</system-info>"
        ),
    ),
)

Use a smaller, faster model for compression by specifying compression_model and compression_formatter to reduce cost and latency.

Structured Output

Pass a Pydantic BaseModel subclass as structured_model when calling the agent. The structured result is available in response.metadata.

from pydantic import BaseModel, Field
from agentscope.message import Msg

class PersonInfo(BaseModel):
    name: str = Field(description="The person's name")
    description: str = Field(description="A one-sentence description")
    age: int = Field(description="The person's age")
    honors: list[str] = Field(description="A list of honors")

response = await agent(
    Msg("user", "Introduce Einstein", "user"),
    structured_model=PersonInfo,
)

print(response.metadata)  # dict with name, description, age, honors

response.get_text_content() still returns the text content. The structured data is in response.metadata.

Planning

The Plan Module enables ReActAgent to formally break down complex tasks into manageable sub-tasks and execute them systematically. Pass a PlanNotebook instance via the plan_notebook parameter to activate it. Once provided, the agent:

Is automatically equipped with plan management tool functions
Receives a hint message at the beginning of each reasoning step guiding it through the current plan

The current plan module requires subtasks to be executed sequentially. Parallel subtask execution is on the roadmap.

Key capabilities:

Creating, modifying, abandoning, and restoring plans
Switching between multiple plans
Gracefully handling interruptions by temporarily suspending the current plan
Real-time visualization and monitoring via plan change hooks

PlanNotebook

PlanNotebook is the core class. It manages plan state, provides tool functions, and generates hint messages.

Name	Type	Description
`max_subtasks`	`int \| None`	Maximum subtasks per plan. Unlimited if `None`.
`plan_to_hint`	`Callable[[Plan \| None], str \| None] \| None`	Generates a hint message from the current plan. Defaults to `DefaultPlanToHint`.
`storage`	`PlanStorageBase \| None`	Storage for historical plans. Defaults to in-memory.

The plan_to_hint callable is the primary interface for prompt engineering. Provide your own implementation for better performance.

PlanStorageBase inherits from StateModule, so plan storage is automatically saved and loaded by session management.

Core attributes and methods:

Type	Name	Description
attribute	`current_plan`	The plan currently being executed
	`storage`	Storage for historical plans
	`plan_to_hint`	Callable that generates hint messages
method	`list_tools`	List all plan management tool functions
	`get_current_hint`	Get the current hint message
	`create_plan`, `view_subtasks`, `revise_current_plan`, `update_subtask_state`, `finish_subtask`, `finish_plan`, `view_historical_plans`, `recover_historical_plan`	Tool functions for managing plans and subtasks
	`register_plan_change_hook`	Register a hook called when the plan changes
	`remove_plan_change_hook`	Remove a registered plan change hook

Manual Plan Specification

Create a plan upfront, then pass the PlanNotebook to ReActAgent:

import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.model import DashScopeChatModel
from agentscope.plan import PlanNotebook, SubTask

plan_notebook = PlanNotebook()

async def setup_plan() -> None:
    await plan_notebook.create_plan(
        name="Research on Agent",
        description="Conduct a comprehensive research on the LLM-empowered agent.",
        expected_outcome=(
            "A Markdown report answering: 1. What's an agent? "
            "2. What's the current state of the art? "
            "3. What's the future trend?"
        ),
        subtasks=[
            SubTask(
                name="Search agent-related survey papers",
                description=(
                    "Search multiple sources including Google Scholar, arXiv, "
                    "and Semantic Scholar. Must be published after 2021 with "
                    "more than 50 citations."
                ),
                expected_outcome="A paper list in Markdown format",
            ),
            SubTask(
                name="Read and summarize the papers",
                description=(
                    "Read the papers and summarize key points: definition, "
                    "taxonomy, challenges, and key directions."
                ),
                expected_outcome="A summary of key points in Markdown format",
            ),
            SubTask(
                name="Research recent advances of large companies",
                description=(
                    "Research recent advances from Google, Microsoft, OpenAI, "
                    "Anthropic, Alibaba, and Meta via official blogs or news articles."
                ),
                expected_outcome="A recent advances summary in Markdown format",
            ),
            SubTask(
                name="Write a report",
                description="Write a report based on all previous steps.",
                expected_outcome=(
                    "A Markdown report answering: 1. What's an agent? "
                    "2. What's the current state of the art? "
                    "3. What's the future trend?"
                ),
            ),
        ],
    )

    msg = await plan_notebook.get_current_hint()
    print(f"{msg.name}: {msg.content}")

asyncio.run(setup_plan())

agent = ReActAgent(
    name="Friday",
    sys_prompt="You are a helpful assistant.",
    model=DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    formatter=DashScopeChatFormatter(),
    plan_notebook=plan_notebook,
)

Agent-Managed Plan Execution

Pass a fresh PlanNotebook and let the agent decide when and how to plan. For complex tasks, the agent will create a plan autonomously and execute it step by step:

from agentscope.agent import UserAgent

agent = ReActAgent(
    name="Friday",
    sys_prompt="You are a helpful assistant.",
    model=DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    formatter=DashScopeChatFormatter(),
    plan_notebook=PlanNotebook(),
)

async def interact_with_agent() -> None:
    user = UserAgent(name="user")
    msg = None
    while True:
        msg = await user(msg)
        if msg.get_text_content() == "exit":
            break
        msg = await agent(msg)

asyncio.run(interact_with_agent())

Plan Visualization and Monitoring

from agentscope.plan import PlanNotebook, Plan

def on_plan_changed(self: PlanNotebook, plan: Plan) -> None:
    """Called whenever the plan is updated.

    Args:
        self: The PlanNotebook instance.
        plan: The updated plan instance.
    """
    # Forward the plan to a frontend or logging system
    ...

plan_notebook.register_plan_change_hook(on_plan_changed)

Customizing Agents

AgentScope provides two base classes for building custom agents:

Class	Abstract Methods	Hooks Supported	Description
`AgentBase`	`reply`, `observe`, `print`, `handle_interrupt`	pre/post reply, observe, print	Base class for all agents
`ReActAgentBase`	All of above + `_reasoning`, `_acting`	All of above + pre/post reasoning, acting	Base for ReAct-style agents
`ReActAgent`	—	All hooks	Production-ready ReAct implementation
`UserAgent`	—	—	Represents a human user in the pipeline
`A2AAgent`	—	pre/post reply, observe, print	Communicates with remote A2A agents

Inherit from AgentBase for simple agents, or ReActAgentBase if you want the reasoning/acting separation with corresponding hooks.

from agentscope.agent import AgentBase
from agentscope.message import Msg

class MyAgent(AgentBase):
    async def reply(self, msg: Msg) -> Msg:
        # your logic here
        return Msg(self.name, "Hello!", "assistant")

    async def handle_interrupt(self, *args, **kwargs) -> Msg:
        return Msg(self.name, "Interrupted.", "assistant")

Agent Hooks

Hooks let you inject custom logic at specific points in an agent’s execution without modifying its core code.

Supported Hook Types

Agent Class	Core Function	Hook Types	Description
`AgentBase` & subclasses	`reply`	`pre_reply`, `post_reply`	Before/after the agent replies
	`print`	`pre_print`, `post_print`	Before/after printing output
	`observe`	`pre_observe`, `post_observe`	Before/after observing a message
`ReActAgentBase` & subclasses	`_reasoning`	`pre_reasoning`, `post_reasoning`	Before/after the reasoning step
	`_acting`	`pre_acting`, `post_acting`	Before/after the acting step

Hooks are implemented via metaclass and support inheritance — subclasses automatically inherit hook support from their parent classes.

Hook Signatures

All pre-hooks share the same signature:

from typing import Any
from agentscope.agent import AgentBase

def my_pre_hook(
    self: AgentBase,
    kwargs: dict[str, Any],
) -> dict[str, Any] | None:
    # modify kwargs and return, or return None to leave them unchanged
    return kwargs

Post-hooks receive an additional output argument:

def my_post_hook(
    self: AgentBase,
    kwargs: dict[str, Any],
    output: Any,
) -> Any | None:
    # modify output and return, or return None to leave it unchanged
    return output

All positional and keyword arguments of the core function are passed as a single kwargs dict. When a hook returns None, the next hook receives the most recent non-None return value (or the original arguments if all previous hooks returned None).

Hook Management

AgentScope provides the following methods to manage instance-level hooks:

Method	Description
`register_instance_hook`	Register a hook for this instance
`remove_instance_hook`	Remove a named hook from this instance
`clear_instance_hooks`	Clear all hooks of a given type from this instance

Example: modifying message content before reply

from agentscope.agent import AgentBase, ReActAgent
from agentscope.message import Msg
from typing import Any

def add_prefix_hook(
    self: AgentBase,
    kwargs: dict[str, Any],
) -> dict[str, Any]:
    msg = kwargs["msg"]
    msg.content = "[reviewed] " + msg.content
    return {**kwargs, "msg": msg}

agent = ReActAgent(...)

agent.register_instance_hook(
    hook_type="pre_reply",
    hook_name="add_prefix",
    hook=add_prefix_hook,
)

# Remove it later
agent.remove_instance_hook("pre_reply", "add_prefix")

Never call the core function (reply, observe, print, _reasoning, _acting) inside a hook — this will cause an infinite loop.

State and Session Management

StateModule

StateModule is the foundation for state management. Any class that inherits from it can register attributes as part of its state, enabling serialization and restoration. AgentBase, MemoryBase, LongTermMemoryBase, and Toolkit all inherit from StateModule.

Method	Description
`register_state(attr_name, custom_to_json?, custom_from_json?)`	Register an attribute as part of the state
`state_dict()`	Get the current state as a dictionary
`load_state_dict(state_dict, strict?)`	Restore state from a dictionary

Attributes that themselves inherit from StateModule are automatically included in the parent’s state (nested serialization):

from agentscope.module import StateModule

class MyMemory(StateModule):
    def __init__(self):
        super().__init__()
        self.messages = []
        self.register_state("messages")

class MyAgent(StateModule):
    def __init__(self):
        super().__init__()
        self.memory = MyMemory()  # auto-included because MyMemory is a StateModule
        self.name = "Friday"
        self.register_state("name")

Saving and restoring agent state:

import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit
from agentscope.message import Msg

agent = ReActAgent(
    name="Friday",
    sys_prompt="You are a helpful assistant.",
    model=DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    formatter=DashScopeChatFormatter(),
    memory=InMemoryMemory(),
    toolkit=Toolkit(),
)

# Save state before any interaction
initial_state = agent.state_dict()

# ... run the agent ...
await agent(Msg("user", "Hello!", "user"))

# Restore to the saved state
agent.load_state_dict(initial_state)

Session Management

A session is a collection of StateModule objects (e.g., multiple agents) whose state you want to persist together. AgentScope provides JSONSession, which saves and loads session state as a JSON file named by session ID:

from agentscope.session import JSONSession

session = JSONSession(save_dir="./sessions")

Saving a session:

await session.save_session_state(
    session_id="user_1",
    agent=agent,          # keyword argument name must match when loading
)

Loading a session:

await session.load_session_state(
    session_id="user_1",
    agent=agent,          # same keyword argument name as used when saving
)

You can pass multiple agents to save_session_state and load_session_state as keyword arguments. The keyword names must be consistent between save and load calls.

JSONSession is a concrete implementation of SessionBase. You can implement your own session class with a custom storage backend (e.g., Redis, a database) by subclassing SessionBase and implementing save_session_state and load_session_state.

A2A Agent

A2A support is an experimental feature and may change in future versions. Current limitations include:

Only supports chatbot scenarios (one user, one agent)
Does not support real-time interruption
Does not support structured output
Messages received via observe are sent to the remote agent only when reply is called

A2AAgent lets you communicate with any remote agent that implements the A2A protocol. The related classes are:

Class	Description
`A2AAgent`	Agent for communicating with remote A2A agents
`A2AChatFormatter`	Converts between AgentScope messages and A2A formats
`FileAgentCardResolver`	Loads an Agent Card from a local JSON file
`WellKnownAgentCardResolver`	Fetches an Agent Card from a remote well-known URL
`NacosAgentCardResolver`	Fetches an Agent Card from a Nacos registry

Obtaining an Agent Card

An Agent Card describes the remote agent’s name, capabilities, and connection details. There are four ways to obtain one. 1. Create manually

from a2a.types import AgentCard, AgentCapabilities

agent_card = AgentCard(
    name="Friday",
    description="A fun chatting companion",
    url="http://localhost:8000",
    version="1.0.0",
    capabilities=AgentCapabilities(
        push_notifications=False,
        state_transition_history=True,
        streaming=True,
    ),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[],
)

2. Fetch from a well-known URL

from agentscope.a2a import WellKnownAgentCardResolver

resolver = WellKnownAgentCardResolver(base_url="http://localhost:8000")
agent_card = await resolver.get_agent_card()

3. Load from a local JSON file

from agentscope.a2a import FileAgentCardResolver

resolver = FileAgentCardResolver(file_path="./agent_card.json")
agent_card = await resolver.get_agent_card()

The JSON file should follow this format:

{
    "name": "RemoteAgent",
    "url": "http://localhost:8000",
    "description": "Remote A2A Agent",
    "version": "1.0.0",
    "capabilities": {},
    "default_input_modes": ["text/plain"],
    "default_output_modes": ["text/plain"],
    "skills": []
}

4. Fetch from Nacos registry

from agentscope.a2a import NacosAgentCardResolver
from v2.nacos import ClientConfig

resolver = NacosAgentCardResolver(
    remote_agent_name="my-remote-agent",
    nacos_client_config=ClientConfig(
        server_addresses="http://localhost:8848",
    ),
)
agent_card = await resolver.get_agent_card()

NacosAgentCardResolver requires a Nacos server version 3.1.0 or higher with the Agent Registry feature enabled.

Using A2AAgent

Once you have an Agent Card, create an A2AAgent and use it like any other agent: Chatbot scenario:

from agentscope.agent import A2AAgent, UserAgent

agent = A2AAgent(agent_card=agent_card)
user = UserAgent("user")

msg = None
while True:
    msg = await user(msg)
    if msg.get_text_content() == "exit":
        break
    msg = await agent(msg)

As a tool function (handoff/router pattern):

from agentscope.agent import A2AAgent
from agentscope.message import Msg, TextBlock
from agentscope.tool import ToolResponse

agent = A2AAgent(agent_card=agent_card)

async def delegate_to_remote(query: str) -> ToolResponse:
    """Complete a task through a remote sub-agent.

    Args:
        query (``str``):
            Description of the task for the sub-agent.
    """
    res = await agent(Msg("user", query, "user"))
    return ToolResponse(
        content=[TextBlock(type="text", text=res.get_text_content())],
    )

Realtime Agent

The realtime agent is currently under active development. Contributions, discussions, and feedback are welcome.

RealtimeAgent is designed for real-time interactions such as voice conversations. It bridges realtime model APIs with your application via a unified event interface.

Supported Providers

Provider	Class	Supported Models	Input Modalities	Tool Support
DashScope	`DashScopeRealtimeModel`	`qwen3-omni-flash-realtime`	Text, Audio, Image	No
OpenAI	`OpenAIRealtimeModel`	`gpt-4o-realtime-preview`	Text, Audio	Yes
Gemini	`GeminiRealtimeModel`	`gemini-2.5-flash-native-audio-preview-09-2025`	Text, Audio, Image	Yes

Initializing a realtime model:

import os
from agentscope.realtime import DashScopeRealtimeModel, OpenAIRealtimeModel, GeminiRealtimeModel

# DashScope
dashscope_model = DashScopeRealtimeModel(
    model_name="qwen3-omni-flash-realtime",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    voice="Cherry",  # "Cherry", "Serena", "Ethan", "Chelsie"
    enable_input_audio_transcription=True,
)

# OpenAI
openai_model = OpenAIRealtimeModel(
    model_name="gpt-4o-realtime-preview",
    api_key=os.getenv("OPENAI_API_KEY"),
    voice="alloy",  # "alloy", "echo", "marin", "cedar"
    enable_input_audio_transcription=True,
)

# Gemini
gemini_model = GeminiRealtimeModel(
    model_name="gemini-2.5-flash-native-audio-preview-09-2025",
    api_key=os.getenv("GEMINI_API_KEY"),
    voice="Puck",  # "Puck", "Charon", "Kore", "Fenrir"
    enable_input_audio_transcription=True,
)

Creating a RealtimeAgent

import asyncio
from agentscope.agent import RealtimeAgent
from agentscope.realtime import DashScopeRealtimeModel

async def main():
    agent = RealtimeAgent(
        name="Friday",
        sys_prompt="You are a helpful assistant named Friday.",
        model=DashScopeRealtimeModel(
            model_name="qwen3-omni-flash-realtime",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
        ),
    )

    outgoing_queue = asyncio.Queue()

    async def handle_events():
        while True:
            event = await outgoing_queue.get()
            print(f"Event: {event.type}")

    asyncio.create_task(handle_events())

    await agent.start(outgoing_queue)
    # ... handle inputs ...
    await agent.stop()

Starting a Realtime Conversation

A typical setup uses a WebSocket server (e.g., FastAPI) as the backend and a browser client as the frontend. Backend (FastAPI):

import asyncio
import os
from fastapi import FastAPI, WebSocket
from agentscope.agent import RealtimeAgent
from agentscope.realtime import DashScopeRealtimeModel, ClientEvents

app = FastAPI()

@app.websocket("/ws/{user_id}/{session_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str, session_id: str):
    await websocket.accept()

    frontend_queue = asyncio.Queue()

    agent = RealtimeAgent(
        name="Assistant",
        sys_prompt="You are a helpful assistant.",
        model=DashScopeRealtimeModel(
            model_name="qwen3-omni-flash-realtime",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
        ),
    )

    await agent.start(frontend_queue)

    async def send_to_frontend():
        while True:
            msg = await frontend_queue.get()
            await websocket.send_json(msg.model_dump())

    asyncio.create_task(send_to_frontend())

    while True:
        data = await websocket.receive_json()
        client_event = ClientEvents.from_json(data)
        await agent.handle_input(client_event)

Frontend (JavaScript):

const ws = new WebSocket('ws://localhost:8000/ws/user1/session1');

ws.onopen = () => {
    ws.send(JSON.stringify({
        type: 'client_session_create',
        config: { instructions: 'You are a helpful assistant.', user_name: 'User1' }
    }));
};

ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    if (data.type === 'response_audio_delta') {
        playAudio(data.delta);
    }
};

function sendAudioChunk(audioData) {
    ws.send(JSON.stringify({
        type: 'client_audio_append',
        session_id: 'session1',
        audio: audioData,  // base64 encoded
        format: { encoding: 'pcm16', sample_rate: 16000 }
    }));
}

Multi-Agent with ChatRoom

ChatRoom manages multiple RealtimeAgent instances in a shared conversation space, with automatic message broadcasting and unified lifecycle management.

import asyncio
import os
from agentscope.pipeline import ChatRoom
from agentscope.agent import RealtimeAgent
from agentscope.realtime import DashScopeRealtimeModel, ClientEvents

async def main():
    agent1 = RealtimeAgent(
        name="Agent1",
        sys_prompt="You are Agent1.",
        model=DashScopeRealtimeModel(
            model_name="qwen3-omni-flash-realtime",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
        ),
    )
    agent2 = RealtimeAgent(
        name="Agent2",
        sys_prompt="You are Agent2.",
        model=DashScopeRealtimeModel(
            model_name="qwen3-omni-flash-realtime",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
        ),
    )

    chat_room = ChatRoom(agents=[agent1, agent2])
    outgoing_queue = asyncio.Queue()

    await chat_room.start(outgoing_queue)

    # Send input — the ChatRoom broadcasts to all agents
    await chat_room.handle_input(
        ClientEvents.ClientTextAppendEvent(
            session_id="session1",
            text="Hello everyone!",
        )
    )

    await chat_room.stop()

Event Reference

ModelEvents (Realtime Model → Agent)

Event	Description
`ModelSessionCreatedEvent`	Session successfully created
`ModelSessionEndedEvent`	Session has ended
`ModelResponseCreatedEvent`	Model begins generating a response
`ModelResponseDoneEvent`	Model finished generating a response
`ModelResponseAudioDeltaEvent`	Streaming audio chunk
`ModelResponseAudioDoneEvent`	Audio response complete
`ModelResponseAudioTranscriptDeltaEvent`	Streaming transcription chunk
`ModelResponseAudioTranscriptDoneEvent`	Audio transcription complete
`ModelResponseToolUseDeltaEvent`	Streaming tool call parameters
`ModelResponseToolUseDoneEvent`	Tool call parameters complete
`ModelInputTranscriptionDeltaEvent`	Streaming user input transcription
`ModelInputTranscriptionDoneEvent`	User input transcription complete
`ModelInputStartedEvent`	Start of user audio input (VAD)
`ModelInputDoneEvent`	End of user audio input (VAD)
`ModelErrorEvent`	An error occurred

ServerEvents (Backend → Frontend)

Event	Description
`ServerSessionCreatedEvent`	Session created in backend
`ServerSessionUpdatedEvent`	Session updated in backend
`ServerSessionEndedEvent`	Session ended in backend
`AgentReadyEvent`	Agent is ready to receive inputs
`AgentEndedEvent`	Agent has ended
`AgentResponseCreatedEvent`	Agent starts generating response
`AgentResponseDoneEvent`	Agent finished generating response
`AgentResponseAudioDeltaEvent`	Streaming audio chunk from agent
`AgentResponseAudioDoneEvent`	Audio response complete
`AgentResponseAudioTranscriptDeltaEvent`	Streaming transcription of agent response
`AgentResponseAudioTranscriptDoneEvent`	Transcription complete
`AgentResponseToolUseDeltaEvent`	Streaming tool call data
`AgentResponseToolUseDoneEvent`	Tool call complete
`AgentResponseToolResultEvent`	Tool execution result
`AgentInputTranscriptionDeltaEvent`	Streaming transcription of user input
`AgentInputTranscriptionDoneEvent`	Input transcription complete
`AgentInputStartedEvent`	User audio input started
`AgentInputDoneEvent`	User audio input ended
`AgentErrorEvent`	An error occurred

ClientEvents (Frontend → Backend)

Event	Description
`ClientSessionCreateEvent`	Create a new session
`ClientSessionEndEvent`	End the current session
`ClientResponseCreateEvent`	Request agent to generate a response immediately
`ClientResponseCancelEvent`	Interrupt the agent’s current response
`ClientTextAppendEvent`	Append text input
`ClientAudioAppendEvent`	Append audio input
`ClientAudioCommitEvent`	Commit audio input (signal end of input)
`ClientImageAppendEvent`	Append image input
`ClientToolResultEvent`	Send tool execution result

ReAct Agent

Customizing Agents

Agent Hooks

State and Session Management

A2A Agent

Realtime Agent

​ReAct Agent

​Realtime Steering

​Memory Compression

​Structured Output

​Planning

​PlanNotebook

​Manual Plan Specification

​Agent-Managed Plan Execution

​Plan Visualization and Monitoring

​Customizing Agents

​Agent Hooks

​Supported Hook Types

​Hook Signatures

​Hook Management

​State and Session Management

​StateModule

​Session Management

​A2A Agent

​Obtaining an Agent Card

​Using A2AAgent

​Realtime Agent

​Supported Providers

​Creating a RealtimeAgent

​Starting a Realtime Conversation

​Multi-Agent with ChatRoom

​Event Reference

ReAct Agent

Realtime Steering

Memory Compression

Structured Output

Planning

PlanNotebook

Manual Plan Specification

Agent-Managed Plan Execution

Plan Visualization and Monitoring

Customizing Agents

Agent Hooks

Supported Hook Types

Hook Signatures

Hook Management

State and Session Management

StateModule

Session Management

A2A Agent

Obtaining an Agent Card

Using A2AAgent

Realtime Agent

Supported Providers

Creating a RealtimeAgent

Starting a Realtime Conversation

Multi-Agent with ChatRoom

Event Reference