> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentscope.io/llms.txt
> Use this file to discover all available pages before exploring further.
# Agent
> A guide to using built-in agents in AgentScope, including ReActAgent, hooks, state management, A2A, and realtime agents.
AgentScope provides several built-in agent types to cover different use cases. Use the cards below to jump to the section you need.
The primary agent for tool-using, reasoning, and structured output tasks.
Build your own agent by extending AgentBase or ReActAgentBase.
Inject custom logic before or after agent core functions.
Save and restore agent state across sessions.
Connect to remote agents using the Agent-to-Agent protocol.
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](/versions/1.0.21/en/building-blocks/tool-capabilities#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](#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.
```python theme={null}
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:
```python theme={null}
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`:
```python theme={null}
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=(
"Summarize the conversation focusing on "
"the main topic, key points, and pending tasks."
),
summary_template=(
"Summary:\n"
"Main Topic: {main_topic}\n\n"
"Key Points:\n{key_points}\n\n"
"Pending Tasks:\n{pending_tasks}"
),
),
)
```
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`.
```python theme={null}
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`:
```python theme={null}
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:
```python theme={null}
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
Register a hook to react whenever the plan changes — useful for forwarding plan state to a frontend or logging system:
```python theme={null}
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.
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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**
```python theme={null}
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):
```python theme={null}
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:**
```python theme={null}
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:
```python theme={null}
from agentscope.session import JSONSession
session = JSONSession(save_dir="./sessions")
```
**Saving a session:**
```python theme={null}
await session.save_session_state(
session_id="user_1",
agent=agent, # keyword argument name must match when loading
)
```
**Loading a session:**
```python theme={null}
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](https://google.github.io/A2A/). 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**
```python theme={null}
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**
```python theme={null}
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**
```python theme={null}
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:
```json theme={null}
{
"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**
```python theme={null}
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:**
```python theme={null}
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):**
```python theme={null}
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:**
```python theme={null}
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
```python theme={null}
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):**
```python theme={null}
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):**
```javascript theme={null}
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.
```python theme={null}
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
| 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 |
| 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 |
| 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 |