games-development-ai/.claude/skills/agentscope-skill/references/v2-building-blocks.md

Agent

Learn how to define and configure agents in AgentScope v2.0

Overview

Agent is the core abstraction in AgentScope — a stateless reasoning-acting loop engine that integrates models, tools, the permission system, human-in-the-loop, context management, middlewares, state management, and the event system into a single unified interface.

Its primary responsibilities are:

• Accepting input messages or events, invoking tools to complete tasks
• Managing context, including context compression and offloading
• Providing middleware hooks at key lifecycle stages for custom logic
• Automatically managing concurrent and sequential tool execution

Core Interfaces

The main interfaces of the Agent class are as follows:

Method	Description
reply(inputs)	Run the reasoning-acting loop and return the final Msg
reply_stream(inputs)	Same as reply, but yields AgentEvent objects as they are produced
observe(msgs)	Add messages to context without triggering reasoning
compress_context(context_config)	Compress the context if token count exceeds the threshold

Main Loop

The agent runs a reasoning-acting loop on every reply call. The diagram below shows the main control flow.

flowchart TD
    A([Input: msg / event]) --> B{Awaiting\noutside event?}
    B -- Yes --> C[Handle Event\nupdate tool states]
    B -- No --> D[Add msgs to context]
    C --> E
    D --> E

    E{Check next action} -- exit --> F([Return: waiting\nfor outside interaction])
    E -- reasoning --> G[Compress context\nif needed]
    G --> H[LLM Call]
    H -- no tool calls --> I([Return final message])
    H -- tool calls --> Acting

    subgraph Acting [Acting]
        direction TB
        J[Batch tool calls\nsequential / concurrent] --> L[Execute tool calls]
        L --> M{Permission\nCheck}
        M -- ALLOW --> N[Run tool → result]
        M -- ASK / External --> O([Pause & emit\nRequireUserConfirmEvent])
        M -- DENY --> P[Error result to LLM]
    end

    N --> E
    P --> E

Configure Agent

Pass parameters to Agent(...) at initialization. The examples below cover the most common setups.

```python Minimal Setup from agentscope.agent import Agent from agentscope.model import DashScopeChatModel from agentscope.credential import DashScopeCredential

agent = Agent( name="my_agent", system_prompt="You are a helpful assistant.", model=DashScopeChatModel( credential=DashScopeCredential(api_key="YOUR_API_KEY"), model="qwen-max", ), )

```python With Tools / MCP / Skills
import os
from agentscope.agent import Agent
from agentscope.tool import Toolkit, Bash, Edit, Grep, Read, Write
from agentscope.mcp import MCPClient, HttpMCPConfig
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential

agent = Agent(
    name="my_agent",
    system_prompt="You are a helpful assistant.",
    model=DashScopeChatModel(
        credential=DashScopeCredential(api_key="YOUR_API_KEY"),
        model="qwen-max",
    ),
    toolkit=Toolkit(
        tools=[Bash(), Edit(), Grep(), Read(), Write()],
        mcps=[
            MCPClient(
                name="amap",
                is_stateful=False,
                mcp_config=HttpMCPConfig(
                    url=f"https://mcp.amap.com/mcp?key={os.environ['AMAP_API_KEY']}",
                ),
            ),
        ],
        skills_or_loaders=["./skills"],
    ),
)

from agentscope.agent import Agent
from agentscope.agent import ContextConfig
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential

agent = Agent(
    name="my_agent",
    system_prompt="You are a helpful assistant.",
    model=DashScopeChatModel(
        credential=DashScopeCredential(api_key="YOUR_API_KEY"),
        model="qwen-max",
    ),
    context_config=ContextConfig(
        trigger_ratio=0.7,       # compress when 70% of context is used
        reserve_ratio=0.2,       # keep the most recent 20% after compression
        tool_result_limit=1000,  # truncate tool results at 1000 tokens
    ),
)

Parameters

Parameter	Type	Default	Description
name	str	required	Agent identifier, used in messages and logs
system_prompt	str	required	The agent's base system prompt
model	ChatModelBase	required	The LLM used for reasoning
toolkit	Toolkit | None	None	Manages tools, MCP clients, skills, and tool groups
state	AgentState | None	auto-created	Holds context, permission context, and session state
offloader	Offloader | None	None	Offloads compressed context and tool results; must implement the Offloader protocol
middlewares	list[MiddlewareBase] | None	None	Applied at reply, reasoning, acting, model call, and system prompt hooks
model_config	ModelConfig	default	Retry count and fallback model
context_config	ContextConfig	default	Context compression thresholds and tool result limits
react_config	ReActConfig	default	Max iterations and rejection handling

Run the Agent

Both reply and reply_stream accept the same inputs parameter and drive the same reasoning-acting loop. The difference is in how results are delivered.

The inputs parameter accepts:

• A single Msg or a list of Msg objects — starts a new reply
• A UserConfirmResultEvent or ExternalExecutionResultEvent — resumes from a paused state
• None — continue from the current state without new input

reply

reply consumes all events internally and returns the final Msg when the agent finishes or pauses for outside interaction.

import asyncio
from agentscope.message import UserMsg

async def main():
    msg = UserMsg(name="user", content="What files are in the current directory?")
    result = await agent.reply(msg)
    print(result.get_text_content())

asyncio.run(main())

reply_stream

reply_stream yields AgentEvent objects as they are produced, letting you stream text output, tool call progress, and lifecycle events to the user in real time.

async def main():
    msg = UserMsg(name="user", content="Summarize the README.")
    async for event in agent.reply_stream(msg):
        if hasattr(event, "delta"):
            print(event.delta, end="", flush=True)

asyncio.run(main())

observe

Use observe to inject messages into the agent's context without triggering a reply — useful in multi-agent setups where one agent needs to see another's output.

await agent.observe(other_agent_msg)

Compress Context

The agent automatically compresses its context when the token count exceeds context_config.trigger_ratio × model.context_length. Compression summarizes older messages and, if an offloader is configured, offloads them to disk.

You can also trigger compression manually:

from agentscope.agent import ContextConfig

# Use the agent's default config
await agent.compress_context()

# Or pass a custom config for this call only
await agent.compress_context(
    ContextConfig(trigger_ratio=0.6, reserve_ratio=0.2)
)

If the system prompt alone exceeds the compression threshold, `compress_context` raises a `RuntimeError`. Keep system prompts concise or increase the model's context length.

Tool results that exceed tool_result_limit tokens are truncated automatically. If an offloader is set, the truncated portion is offloaded and the agent receives a path reference it can read on demand.

Human-in-the-Loop

The agent pauses execution and emits special events when it encounters two situations: a tool call that requires user confirmation (permission system returns ASK), or a tool marked as external execution (the result must come from outside the agent). In both cases, you resume the agent by passing a result event back via reply.

User Confirmation

When the permission system determines a tool call needs user approval, the agent emits a RequireUserConfirmEvent and pauses.

Use `reply_stream` to detect the pause. The event has the following structure:

<ParamField path="reply_id" type="str" required>
  ID of the current reply, used to resume the agent.
</ParamField>

<ParamField path="tool_calls" type="list[ToolCallBlock]" required>
  Tool calls pending user confirmation. Each `ToolCallBlock` contains:

  <Expandable>
    <ParamField path="id" type="str">Unique identifier for this tool call.</ParamField>
    <ParamField path="name" type="str">The tool name (e.g. `"Bash"`, `"Write"`).</ParamField>
    <ParamField path="input" type="str">JSON-encoded input parameters.</ParamField>
    <ParamField path="suggested_rules" type="list[PermissionRule]">Auto-generated permission rules the user can accept to allow similar future calls.</ParamField>
  </Expandable>
</ParamField>

```python
from agentscope.event import RequireUserConfirmEvent

async for event in agent.reply_stream(msg):
    if isinstance(event, RequireUserConfirmEvent):
        for tc in event.tool_calls:
            print(f"Tool: {tc.name}, Input: {tc.input}")
            print(f"Suggested rules: {tc.suggested_rules}")
```

For each pending tool call, create a `ConfirmResult` indicating whether to allow or deny it. You can also modify the tool call input or accept suggested permission rules:

```python
from agentscope.event import ConfirmResult, UserConfirmResultEvent

confirm_results = []
for tc in event.tool_calls:
    confirm_results.append(ConfirmResult(
        confirmed=True,           # or False to deny
        tool_call=tc,             # pass back (optionally modified)
        rules=tc.suggested_rules, # accept rules for future auto-allow
    ))
```

Pass the `UserConfirmResultEvent` back to `reply` or `reply_stream`:

```python
confirm_event = UserConfirmResultEvent(
    reply_id=event.reply_id,
    confirm_results=confirm_results,
)
result = await agent.reply(confirm_event)
```

* **Confirmed** tool calls execute immediately and the agent continues reasoning
* **Denied** tool calls produce an error result visible to the LLM, which may retry with a different approach
* **Accepted rules** are persisted to the permission engine — matching future calls are auto-allowed without prompting again

External Tool Execution

When the agent calls a tool with is_external_tool = True, it emits a RequireExternalExecutionEvent and pauses. The tool's logic runs outside the agent — typically by a human operator or an external system.

The event has the following structure:

<ParamField path="reply_id" type="str" required>
  ID of the current reply, used to resume the agent.
</ParamField>

<ParamField path="tool_calls" type="list[ToolCallBlock]" required>
  Tool calls to be executed externally. Each `ToolCallBlock` contains:

  <Expandable>
    <ParamField path="id" type="str">Unique identifier for this tool call.</ParamField>
    <ParamField path="name" type="str">The external tool name.</ParamField>
    <ParamField path="input" type="str">JSON-encoded input parameters.</ParamField>
  </Expandable>
</ParamField>

```python
from agentscope.event import RequireExternalExecutionEvent

async for event in agent.reply_stream(msg):
    if isinstance(event, RequireExternalExecutionEvent):
        for tc in event.tool_calls:
            print(f"Execute externally: {tc.name}({tc.input})")
```

Run the operation outside the agent and wrap the results as `ToolResultBlock` objects:

```python
from agentscope.message import ToolResultBlock, TextBlock, ToolResultState
from agentscope.event import ExternalExecutionResultEvent

execution_results = []
for tc in event.tool_calls:
    # Perform the actual operation (API call, human action, etc.)
    output = await run_external_operation(tc.name, tc.input)

    execution_results.append(ToolResultBlock(
        id=tc.id,
        name=tc.name,
        output=[TextBlock(text=output)],
        state=ToolResultState.SUCCESS,
    ))
```

Pass the `ExternalExecutionResultEvent` back to resume:

```python
external_event = ExternalExecutionResultEvent(
    reply_id=event.reply_id,
    execution_results=execution_results,
)
result = await agent.reply(external_event)
```

The results are injected into the agent's context and reasoning continues from where it left off.

Use `reply_stream` when building interactive UIs — it lets you detect pause events in real time and prompt the user immediately. Use `reply` when you have pre-built automation that handles events programmatically.

Persist Agent State

AgentState is a Pydantic model that holds everything needed to resume an agent exactly where it left off — conversation context, compression summary, permission rules, tool state, and the current reply position. Because it is a plain Pydantic model, it serialises to JSON and can be stored in any backend.

RedisStorage is the built-in storage backend. It organises state under a (user_id, agent_id, session_id) key hierarchy and exposes two focused methods for the hot path:

Method	Description
get_session(user_id, agent_id, session_id)	Load a SessionRecord whose .state field is the saved AgentState
update_session_state(user_id, agent_id, session_id, state)	Persist the updated AgentState back to Redis after a reply

import asyncio
from agentscope.agent import Agent
from agentscope.state import AgentState
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg
from agentscope.app.storage import RedisStorage

USER_ID = "user_123"
AGENT_ID = "agent_456"
SESSION_ID = "session_789"

async def main():
    async with RedisStorage(host="localhost", port=6379) as storage:
        # Load state from storage, fall back to a fresh state if not found
        record = await storage.get_session(
            user_id=USER_ID,
            agent_id=AGENT_ID,
            session_id=SESSION_ID,
        )
        state = record.state if record else AgentState()

        # Create the agent with the restored state
        agent = Agent(
            name="my_agent",
            system_prompt="You are a helpful assistant.",
            model=DashScopeChatModel(
                credential=DashScopeCredential(api_key="YOUR_API_KEY"),
                model="qwen-max",
            ),
            state=state,
        )

        # Run a reply turn
        result = await agent.reply(
            UserMsg(name="user", content="Continue where we left off."),
        )
        print(result.get_text_content())

        # Persist the updated state back to Redis
        await storage.update_session_state(
            user_id=USER_ID,
            agent_id=AGENT_ID,
            session_id=SESSION_ID,
            state=agent.state,
        )

asyncio.run(main())

`update_session_state` raises `KeyError` if the session does not exist yet. Use `upsert_session` to create the session record on the first turn, then switch to `update_session_state` for subsequent turns.

Further Reading

Control which tools the agent can call and under what conditions. Intercept and modify agent behavior at reply, reasoning, acting, and model call hooks.

Model

Configure and connect LLM model providers in AgentScope

Overview

The model layer is organized as a two-tier hierarchy: a Credential at the top, and the model families a provider exposes beneath it — Chat Model, TTS, Embedding, and Realtime Model.

  <Tree.File name="AnthropicChatModel" />

  <Tree.File name="DashScopeChatModel" />

  <Tree.File name="..." />
</Tree.Folder>

<Tree.Folder name="TTSModelBase (coming soon)" />

<Tree.Folder name="EmbeddingModelBase (coming soon)" />

<Tree.Folder name="RealtimeModelBase (coming soon)" />

</Tree.Folder>

A Credential carries the API authentication fields a provider requires (api_key, base_url, ...). From a credential, you can retrieve the list of available models for each model family that provider supports.

This layering mirrors the natural frontend flow — register a credential first, then pick a model from under it — letting the UI authenticate once and surface every model family the provider supports.

Chat Model

A Chat Model is the LLM that drives an agent's conversation and tool calls, accepting and producing multimodal content beyond plain text. AgentScope currently ships the following chat model classes:

Provider	Model Class	Highlights
OpenAI	OpenAIChatModel	Chat Completions API, compatible with vLLM and OpenAI-compatible endpoints
OpenAI (Responses)	OpenAIResponseModel	Responses API with native reasoning support (o3, o4-mini)
Anthropic	AnthropicChatModel	Claude models with extended thinking and prompt caching
DashScope	DashScopeChatModel	Qwen models, multimodal (vision/audio/video), reasoning
DeepSeek	DeepSeekChatModel	OpenAI-compatible with prompt cache hit tokens
Gemini	GeminiChatModel	Google Gemini models with multimodal support
Moonshot	MoonshotChatModel	Kimi models (OpenAI-compatible)
xAI	XAIChatModel	Grok models with native reasoning effort
Ollama	OllamaChatModel	Local LLM hosting, credential is optional

Create Chat Model

Every chat model takes a credential, a model name, and an optional provider-specific Parameters object. The three tabs below show typical setups for streaming, tool calling, and reasoning:

```python Streaming import os from agentscope.model import DashScopeChatModel from agentscope.credential import DashScopeCredential

model = DashScopeChatModel( credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]), model="qwen-plus", stream=True, )

```python Tools
import os
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential

model = DashScopeChatModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="qwen-plus",
    stream=False,
    parameters=DashScopeChatModel.Parameters(
        parallel_tool_calls=False,
    ),
)

import os
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential

model = DashScopeChatModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="qwen3-235b-a22b-thinking-2507",
    parameters=DashScopeChatModel.Parameters(
        thinking_enable=True,
        thinking_budget=2048,
    ),
)

Common constructor arguments shared by every chat model:

Argument	Type	Description
credential	CredentialBase	Provider-specific credential
model	str	Model identifier (e.g. "qwen-plus")
parameters	Parameters | None	Provider-specific parameters such as temperature, thinking_enable, parallel_tool_calls
stream	bool	Whether to stream output
max_retries	int	Maximum API retries on failure
context_size	int	Context window used for context compression
formatter	FormatterBase | None	Override message formatter

Call Chat Model

Invoke the model by calling it with a list of Msg objects, plus optional tools and tool_choice:

async def __call__(
    self,
    messages: list[Msg],
    tools: list[dict] | None = None,
    tool_choice: ToolChoice | None = None,
    **kwargs: Any,
) -> ChatResponse | AsyncGenerator[ChatResponse, None]:

The return type follows the model's stream setting:

• stream=False — awaits a single ChatResponse carrying the full output.
• stream=True — awaits an AsyncGenerator[ChatResponse, None]. Intermediate chunks (is_last=False) carry only the delta generated in that step. So that callers don't have to accumulate deltas themselves, AgentScope appends one final chunk with is_last=True that carries the full accumulated content.

import asyncio
import os
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg

async def main():
    model = DashScopeChatModel(
        credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
        model="qwen-plus",
        stream=True,
    )
    msgs = [UserMsg(name="user", content="Count from 1 to 5.")]

    async for chunk in await model(msgs):
        if chunk.is_last:
            print("Final:", chunk.content)   # full accumulated content
        else:
            print("Delta:", chunk.content)   # delta only

asyncio.run(main())

A representative streaming trace, illustrating the delta-then-accumulated pattern:

Delta: [TextBlock(text='1')]
Delta: [TextBlock(text=', 2,')]
Delta: [TextBlock(text=' 3, ')]
Delta: [TextBlock(text='4, 5')]
Final: [TextBlock(text='1, 2, 3, 4, 5')]

Each ChatResponse carries content blocks (TextBlock, ThinkingBlock, ToolCallBlock, DataBlock), an is_last flag, and a ChatUsage recording token counts and elapsed time.

Generate Structured Output

When you need a JSON object that conforms to a Pydantic model or JSON schema, call generate_structured_output instead of __call__. It returns a StructuredResponse whose content is a validated dict matching the schema:

import asyncio
import os
from pydantic import BaseModel
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg

class WeatherInfo(BaseModel):
    city: str
    temperature: float
    unit: str

async def main():
    model = DashScopeChatModel(
        credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
        model="qwen-plus",
        stream=False,
    )
    response = await model.generate_structured_output(
        messages=[UserMsg(name="user", content="What's the weather in Shanghai?")],
        structured_model=WeatherInfo,
    )
    print(response.content)  # validated dict matching WeatherInfo

asyncio.run(main())

`generate_structured_output` synthesizes a forced tool call from the schema, then validates and repairs the model's response.

Formatter

A formatter translates AgentScope's Msg objects into the list[dict] payload that each provider's API expects. It is configured via the optional formatter argument on the chat model constructor. Every provider ships two built-in variants:

Variant	Use Case
ChatFormatter (default)	Standard single-agent dialog. Each Msg maps 1:1 to an API message, preserving native roles (user, assistant, system).
MultiAgentFormatter	Multi-agent scenarios such as debate or moderation. Consecutive agent messages are grouped and wrapped in <history> tags with the sender's name, while tool call / result sequences keep their native API format.

Switch to multi-agent mode by passing the MultiAgent variant — no agent code changes are required:

import os
from agentscope.model import OpenAIChatModel
from agentscope.credential import OpenAICredential
from agentscope.formatter import OpenAIMultiAgentFormatter

model = OpenAIChatModel(
    credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
    model="gpt-4.1",
    formatter=OpenAIMultiAgentFormatter(),
)

For non-standard payload shapes (e.g. a provider whose API doesn't follow the OpenAI or Anthropic conventions), subclass FormatterBase and pass an instance through the same formatter argument.

Custom Provider

You can extend AgentScope with your own model provider by implementing a credential and a chat model, then registering the credential.

Step 1: Define the Credential

Subclass CredentialBase with a unique type discriminator and implement get_chat_model_class():

from typing import Literal, Type, TYPE_CHECKING
from pydantic import ConfigDict, Field, SecretStr
from agentscope.credential import CredentialBase

if TYPE_CHECKING:
    from agentscope.model import ChatModelBase

class MyProviderCredential(CredentialBase):
    model_config = ConfigDict(title="My Provider API")
    type: Literal["my_provider_credential"] = "my_provider_credential"

    api_key: SecretStr = Field(description="API key for My Provider.")
    base_url: str = Field(default="https://api.myprovider.com/v1")

    @classmethod
    def get_chat_model_class(cls) -> Type["ChatModelBase"]:
        from .my_model import MyProviderChatModel
        return MyProviderChatModel

Step 2: Implement the Chat Model

Subclass ChatModelBase, define a Parameters inner class, and implement _call_api:

from typing import Literal, Any, AsyncGenerator
from pydantic import BaseModel, Field
from agentscope.model import ChatModelBase, ChatResponse
from agentscope.message import Msg
from agentscope.tool import ToolChoice
from agentscope.formatter import FormatterBase, OpenAIChatFormatter

class MyProviderChatModel(ChatModelBase):
    class Parameters(BaseModel):
        max_tokens: int | None = Field(default=None, gt=0)
        temperature: float | None = Field(default=None, ge=0, le=2)

    type: Literal["my_provider_chat"] = "my_provider_chat"

    def __init__(
        self,
        credential: "MyProviderCredential",
        model: str,
        parameters: Parameters | None = None,
        stream: bool = True,
        max_retries: int = 3,
        context_size: int = 128000,
        formatter: FormatterBase | None = None,
    ) -> None:
        super().__init__(
            credential=credential,
            model=model,
            parameters=parameters or self.Parameters(),
            stream=stream,
            max_retries=max_retries,
            context_size=context_size,
        )
        # If your API follows the OpenAI format, reuse OpenAIChatFormatter;
        # otherwise implement your own FormatterBase subclass.
        self.formatter = formatter or OpenAIChatFormatter()

    async def _call_api(
        self,
        model_name: str,
        messages: list[Msg],
        tools: list[dict] | None = None,
        tool_choice: ToolChoice | None = None,
        **kwargs: Any,
    ) -> ChatResponse | AsyncGenerator[ChatResponse, None]:
        formatted_messages = await self.formatter.format(messages)
        # Call your provider's API using self.credential.api_key, etc.
        ...

Step 3: Add Model Cards (optional)

Drop YAML files into a _models/ directory next to your model implementation. Each file describes one model — its capabilities (input_types, output_types), limits (context_size, output_size), and any per-model parameter_overrides:

name: my-model-v1
label: My Model V1
status: active
input_types:
  - text/plain
output_types:
  - text/plain
context_size: 128000
output_size: 4096
parameter_overrides:
  max_tokens: {"maximum": 4096}

MyProviderChatModel.list_models() then loads every YAML in that directory. To pull cards from a different location — for example, a registry your application maintains separately — pass custom_yaml_dir:

cards = MyProviderChatModel.list_models(custom_yaml_dir="/path/to/cards")

Integrate with Frontend

What is ModelCard

ModelCard is a declarative description of a model's capabilities and constraints, designed to drive the frontend — model selectors, parameter forms, and feature toggles can be rendered dynamically without hardcoding any provider-specific knowledge.

Each ModelCard contains:

Field	Type	Description
name	str	Model identifier (e.g. "claude-sonnet-4-6")
label	str	Human-readable display name (e.g. "Claude Sonnet 4.6")
status	"active" | "deprecated" | "sunset"	Model lifecycle status
input_types	list[str]	Accepted input MIME types — used by the frontend to filter attachment uploads (e.g. only show an image button when image/* is supported)
output_types	list[str]	Output MIME types the model can produce — advertises capabilities such as a thinking toggle when application/x-thinking is present
context_size	int	Maximum context window in tokens
output_size	int	Maximum output tokens
parameter_schema	dict	Final JSON Schema for the parameter form — base schema merged with per-model overrides (see below)
parameters_overrides	dict[str, dict]	The raw per-model overrides, before merging

input_types and output_types use MIME types to describe modality. Common values:

MIME Type	Meaning
text/plain	Text
application/x-thinking	Reasoning / chain-of-thought
image/* (e.g. image/png, image/jpeg)	Image
audio/* (e.g. audio/wav, audio/mp3)	Audio
video/* (e.g. video/mp4)	Video

A typical YAML card for claude-sonnet-4-6:

name: claude-sonnet-4-6
label: Claude Sonnet 4.6
status: active

input_types:
  - text/plain
  - image/jpeg
  - image/png
  - image/gif
  - image/webp

output_types:
  - text/plain
  - application/x-thinking

context_size: 1000000
output_size: 65536

parameter_overrides:
  max_tokens: {"maximum": 65536}

Parameter schema and overrides

The parameter_schema exposed to the frontend is built in two layers:

1. Base schema — auto-derived from the chat model's Parameters class via model_json_schema(). This lists every adjustable parameter (temperature, max_tokens, thinking_enable, ...) along with its type and the API-wide range.
2. Per-model overrides — the YAML's parameter_overrides block is merged on top, field by field.

Overrides matter because adjustable ranges are not uniform across an API: every Qwen model accepts max_tokens, but each one has a different ceiling. Overrides let a card tighten a range, pin a default, or hide a parameter that doesn't apply.

Override syntax	Effect
param: { ... }	Shallow-merge into the base field (e.g. max_tokens: {maximum: 16384})
param: { hidden: true }	Hide the parameter from the frontend
param: null	Remove the parameter entirely

Retrieve ModelCards

You retrieve model cards by calling list_models() on either the credential class or the model class. Internally, CredentialBase.list_models() delegates to its linked ChatModelBase subclass (obtained via get_chat_model_class()), which loads YAML card definitions from its _models/ directory.

from agentscope.credential import DashScopeCredential
from agentscope.model import AnthropicChatModel

# Via credential class
cards = DashScopeCredential.list_models()

# Or directly on the model class
cards = AnthropicChatModel.list_models()

for card in cards:
    print(f"{card.name}: context={card.context_size}, inputs={card.input_types}")

The credential's get_chat_model_class() returns the corresponding ChatModelBase subclass, which in turn knows where to find its model card YAML files:

model_cls = DashScopeCredential.get_chat_model_class()  # -> DashScopeChatModel
cards = model_cls.list_models()                          # -> list[ModelCard]

This design allows the frontend to discover available models, their capabilities, and valid parameter ranges — all from a single credential, without any hardcoded provider logic.

TTS

Coming soon — we are migrating TTS support from v1.0 to v2.0.

Embedding

Coming soon — we are migrating Embedding support from v1.0 to v2.0.

Realtime Model

Coming soon — we are migrating Realtime Model support from v1.0 to v2.0.

Message & Event

The core data abstractions for agent communication and streaming

Message and Event are the two fundamental data structures in AgentScope.

• Message — the unit of inter-agent communication and persistence. Each Msg represents a complete conversation turn that is stored in context and exchanged between agents.
• Event — the unit of frontend interaction and streaming. Events carry incremental progress updates (text tokens, tool call fragments, permission requests) and drive real-time UIs and human-in-the-loop workflows.

A sequence of events produced during a single reply call accumulates into exactly one assistant Msg. This guarantees that the complete message state is always recoverable from its event stream.

Message

A Msg represents a single turn in a conversation — a user input, an assistant response, or a system instruction, carrying structured content as a list of typed blocks.

One assistant message corresponds to one complete `reply` cycle of the agent (repeated reasoning and acting until a final response is produced).

Structure

The Msg class has the following core fields:

Field	Type	Description
id	str	Unique message identifier
name	str	Name of the sender
role	"user" | "assistant" | "system"	The sender's role
content	list[ContentBlock]	Ordered list of content blocks
metadata	dict	Arbitrary key-value metadata
created_at	str	ISO 8601 timestamp of creation
finished_at	str | None	ISO 8601 timestamp when the message was finalized
usage	Usage	Token usage statistics (for assistant messages)

Content Blocks

Message content is composed of typed blocks. Each block represents a distinct piece of information:

Block	Description	Allowed In
TextBlock	Plain text content	user, assistant, system
DataBlock	Binary data (images, audio) via base64 or URL	user, assistant
ThinkingBlock	Model reasoning (chain-of-thought)	assistant
ToolCallBlock	A tool invocation with name, input, and state	assistant
ToolResultBlock	The output of a tool execution	assistant
HintBlock	Out-of-band hint injected into the conversation (e.g. a scheduled-task trigger, a team message, a background-tool result). The hint field is str for plain text or list[TextBlock | DataBlock] for multimodal payloads; source carries a small JSON tag the frontend uses to label the hint's origin.	assistant

Role constraints are enforced at construction: `user` messages can only contain `TextBlock` and `DataBlock`; `system` messages can only contain `TextBlock`; `assistant` messages can contain all block types.

Create Messages

AgentScope provides three shortcut factory functions for quickly creating messages with the correct role, without manually specifying role or wrapping content into blocks:

Factory	Role	Allowed Content
UserMsg(name, content)	user	str or list[TextBlock | DataBlock]
AssistantMsg(name, content)	assistant	str or list[ContentBlock]
SystemMsg(name, content)	system	str or list[TextBlock]

When content is a plain string, it is automatically wrapped into a TextBlock.

from agentscope.message import UserMsg, AssistantMsg, SystemMsg

# User message — text and optional images
user_msg = UserMsg(name="user", content="What's in this image?")

# User message with multimodal content
from agentscope.message import TextBlock, DataBlock, Base64Source
user_msg = UserMsg(
    name="user",
    content=[
        TextBlock(text="Describe this image:"),
        DataBlock(source=Base64Source(data="...", media_type="image/png")),
    ],
)

# System message — text only
system_msg = SystemMsg(name="system", content="You are a helpful assistant.")

# Assistant message — all block types allowed
assistant_msg = AssistantMsg(name="agent", content="Here is the result...")

Access Content

Msg provides helper methods to extract specific block types:

Method	Returns
get_text_content(separator="\n")	Concatenated text from all TextBlocks, or None
get_content_blocks(block_type)	Filtered list of blocks by type
has_content_blocks(block_type)	True if blocks of the given type exist

# Get all text content
text = msg.get_text_content()

# Get all tool calls
tool_calls = msg.get_content_blocks("tool_call")

# Check if message has tool results
if msg.has_content_blocks("tool_result"):
    ...

Event

Events are the streaming counterpart of messages. While the agent executes, it yields a sequence of AgentEvent objects that represent incremental progress — text tokens arriving, tool calls being constructed, results streaming back. Each event is lightweight and self-contained.

Event Lifecycle

Every event carries a reply_id that links it to the message being constructed. Within a reply, block_id or tool_call_id identifies which content block an event belongs to. Events follow a start → delta → end pattern for each content block:

sequenceDiagram
    participant Client
    participant Agent

    Agent->>Client: ReplyStartEvent

    rect rgba(100, 150, 255, 0.1)
        Note over Client,Agent: Reasoning Phase
        Agent->>Client: ModelCallStartEvent
        rect rgba(200, 200, 100, 0.1)
            Note over Client,Agent: TextBlock (block_id)
            Agent->>Client: TextBlockStartEvent
            Agent->>Client: TextBlockDeltaEvent (×N)
            Agent->>Client: TextBlockEndEvent
        end
        rect rgba(200, 200, 100, 0.1)
            Note over Client,Agent: DataBlock (block_id)
            Agent->>Client: DataBlockStartEvent
            Agent->>Client: DataBlockDeltaEvent (×N)
            Agent->>Client: DataBlockEndEvent
        end
        rect rgba(200, 200, 100, 0.1)
            Note over Client,Agent: ToolCallBlock (tool_call_id)
            Agent->>Client: ToolCallStartEvent
            Agent->>Client: ToolCallDeltaEvent (×N)
            Agent->>Client: ToolCallEndEvent
        end
        Agent->>Client: ModelCallEndEvent
    end

    rect rgba(100, 255, 150, 0.1)
        Note over Client,Agent: Acting Phase
        rect rgba(200, 200, 100, 0.1)
            Note over Client,Agent: ToolResultBlock (tool_call_id)
            Agent->>Client: ToolResultStartEvent
            Agent->>Client: ToolResultTextDeltaEvent (×N)
            Agent->>Client: ToolResultDataDeltaEvent (×N)
            Agent->>Client: ToolResultEndEvent
        end
    end

    Agent->>Client: ReplyEndEvent

All events within the same reply share the same reply_id. Within a reply, use block_id to correlate text/thinking/data block events, and tool_call_id to correlate tool call and tool result events.

Event Types

All events inherit from EventBase which provides common fields:

Field	Type	Description
id	str	Unique event identifier
created_at	str	ISO 8601 timestamp

Events are grouped by category below. Every event also carries a reply_id field (except where noted) that links it to the message being constructed.

**ReplyStartEvent** — Agent begins a new reply.

| Field        | Type  | Description                        |
| ------------ | ----- | ---------------------------------- |
| `reply_id`   | `str` | ID of the reply message            |
| `session_id` | `str` | ID of the session                  |
| `name`       | `str` | Agent name                         |
| `role`       | `str` | Agent role (default `"assistant"`) |

**ReplyEndEvent** — Agent finishes the reply.

| Field        | Type  | Description             |
| ------------ | ----- | ----------------------- |
| `reply_id`   | `str` | ID of the reply message |
| `session_id` | `str` | ID of the session       |

**ExceedMaxItersEvent** — Agent reached the maximum reasoning-acting iterations.

| Field      | Type  | Description             |
| ---------- | ----- | ----------------------- |
| `reply_id` | `str` | ID of the reply message |
| `name`     | `str` | Agent name              |

**TextBlockStartEvent** — A new text block begins.

| Field      | Type  | Description                         |
| ---------- | ----- | ----------------------------------- |
| `reply_id` | `str` | ID of the reply message             |
| `block_id` | `str` | Unique identifier of the text block |

**TextBlockDeltaEvent** — Incremental text content arrives.

| Field      | Type  | Description                         |
| ---------- | ----- | ----------------------------------- |
| `reply_id` | `str` | ID of the reply message             |
| `block_id` | `str` | Unique identifier of the text block |
| `delta`    | `str` | Incremental text content            |

**TextBlockEndEvent** — The text block is complete.

| Field      | Type  | Description                         |
| ---------- | ----- | ----------------------------------- |
| `reply_id` | `str` | ID of the reply message             |
| `block_id` | `str` | Unique identifier of the text block |

**ThinkingBlockStartEvent** — A new thinking block begins.

| Field      | Type  | Description                             |
| ---------- | ----- | --------------------------------------- |
| `reply_id` | `str` | ID of the reply message                 |
| `block_id` | `str` | Unique identifier of the thinking block |

**ThinkingBlockDeltaEvent** — Incremental thinking content arrives.

| Field      | Type  | Description                             |
| ---------- | ----- | --------------------------------------- |
| `reply_id` | `str` | ID of the reply message                 |
| `block_id` | `str` | Unique identifier of the thinking block |
| `delta`    | `str` | Incremental thinking text               |

**ThinkingBlockEndEvent** — The thinking block is complete.

| Field      | Type  | Description                             |
| ---------- | ----- | --------------------------------------- |
| `reply_id` | `str` | ID of the reply message                 |
| `block_id` | `str` | Unique identifier of the thinking block |

**DataBlockStartEvent** — A new data block begins (image, audio, etc.).

| Field        | Type  | Description                         |
| ------------ | ----- | ----------------------------------- |
| `reply_id`   | `str` | ID of the reply message             |
| `block_id`   | `str` | Unique identifier of the data block |
| `media_type` | `str` | MIME type (e.g. `"image/png"`)      |

**DataBlockDeltaEvent** — Incremental binary data arrives.

| Field        | Type  | Description                         |
| ------------ | ----- | ----------------------------------- |
| `reply_id`   | `str` | ID of the reply message             |
| `block_id`   | `str` | Unique identifier of the data block |
| `data`       | `str` | Incremental base64-encoded data     |
| `media_type` | `str` | MIME type                           |

**DataBlockEndEvent** — The data block is complete.

| Field      | Type  | Description                         |
| ---------- | ----- | ----------------------------------- |
| `reply_id` | `str` | ID of the reply message             |
| `block_id` | `str` | Unique identifier of the data block |

**ToolCallStartEvent** — The agent begins a tool call.

| Field            | Type  | Description                        |
| ---------------- | ----- | ---------------------------------- |
| `reply_id`       | `str` | ID of the reply message            |
| `tool_call_id`   | `str` | Unique identifier of the tool call |
| `tool_call_name` | `str` | Name of the tool being called      |

**ToolCallDeltaEvent** — Incremental tool call input arrives.

| Field          | Type  | Description                             |
| -------------- | ----- | --------------------------------------- |
| `reply_id`     | `str` | ID of the reply message                 |
| `tool_call_id` | `str` | Unique identifier of the tool call      |
| `delta`        | `str` | Incremental JSON fragment of tool input |

**ToolCallEndEvent** — The tool call input is complete.

| Field          | Type  | Description                        |
| -------------- | ----- | ---------------------------------- |
| `reply_id`     | `str` | ID of the reply message            |
| `tool_call_id` | `str` | Unique identifier of the tool call |

**ToolResultStartEvent** — Tool execution begins.

| Field            | Type  | Description                       |
| ---------------- | ----- | --------------------------------- |
| `reply_id`       | `str` | ID of the reply message           |
| `tool_call_id`   | `str` | ID of the corresponding tool call |
| `tool_call_name` | `str` | Name of the tool                  |

**ToolResultTextDeltaEvent** — Incremental text output from the tool.

| Field          | Type  | Description                       |
| -------------- | ----- | --------------------------------- |
| `reply_id`     | `str` | ID of the reply message           |
| `tool_call_id` | `str` | ID of the corresponding tool call |
| `delta`        | `str` | Incremental text content          |

**ToolResultDataDeltaEvent** — Binary data output from the tool.

| Field          | Type          | Description                                                  |
| -------------- | ------------- | ------------------------------------------------------------ |
| `reply_id`     | `str`         | ID of the reply message                                      |
| `tool_call_id` | `str`         | ID of the corresponding tool call                            |
| `block_id`     | `str`         | Unique identifier of the data block                          |
| `media_type`   | `str`         | MIME type of the content                                     |
| `data`         | `str \| None` | Base64-encoded data (mutually exclusive with `url`)          |
| `url`          | `str \| None` | URL pointing to the content (mutually exclusive with `data`) |

**ToolResultEndEvent** — Tool execution is complete.

| Field          | Type              | Description                                                            |
| -------------- | ----------------- | ---------------------------------------------------------------------- |
| `reply_id`     | `str`             | ID of the reply message                                                |
| `tool_call_id` | `str`             | ID of the corresponding tool call                                      |
| `state`        | `ToolResultState` | Final state: `SUCCESS`, `ERROR`, `INTERRUPTED`, `DENIED`, or `RUNNING` |

**ModelCallStartEvent** — A model API call begins.

| Field        | Type  | Description                    |
| ------------ | ----- | ------------------------------ |
| `reply_id`   | `str` | ID of the reply message        |
| `model_name` | `str` | Name of the model being called |

**ModelCallEndEvent** — A model API call completes.

| Field           | Type  | Description                       |
| --------------- | ----- | --------------------------------- |
| `reply_id`      | `str` | ID of the reply message           |
| `input_tokens`  | `int` | Number of input tokens consumed   |
| `output_tokens` | `int` | Number of output tokens generated |

**RequireUserConfirmEvent** — Agent pauses for user confirmation.

| Field        | Type                  | Description                          |
| ------------ | --------------------- | ------------------------------------ |
| `reply_id`   | `str`                 | ID of the reply message              |
| `tool_calls` | `list[ToolCallBlock]` | Tool calls pending user confirmation |

**RequireExternalExecutionEvent** — Agent pauses for external execution.

| Field        | Type                  | Description                          |
| ------------ | --------------------- | ------------------------------------ |
| `reply_id`   | `str`                 | ID of the reply message              |
| `tool_calls` | `list[ToolCallBlock]` | Tool calls to be executed externally |

**UserConfirmResultEvent** — User provides confirmation results (input event).

| Field             | Type                  | Description                                     |
| ----------------- | --------------------- | ----------------------------------------------- |
| `reply_id`        | `str`                 | ID of the reply message                         |
| `confirm_results` | `list[ConfirmResult]` | Confirmation results for each pending tool call |

**ExternalExecutionResultEvent** — External system provides execution results (input event).

| Field               | Type                    | Description                               |
| ------------------- | ----------------------- | ----------------------------------------- |
| `reply_id`          | `str`                   | ID of the reply message                   |
| `execution_results` | `list[ToolResultBlock]` | Results returned by the external executor |

Unlike text / thinking / data / tool blocks, these events do not follow the start → delta → end pattern. The full payload arrives in a single event because it is known up-front rather than streamed.

**HintBlockEvent** — A `HintBlock` is injected into the agent's context (e.g. a scheduled-task trigger, a team message, a result returned by an offloaded background tool).

| Field      | Type                                  | Description                                                                                                     |
| ---------- | ------------------------------------- | --------------------------------------------------------------------------------------------------------------- |
| `reply_id` | `str`                                 | ID of the reply message                                                                                         |
| `block_id` | `str`                                 | Unique identifier of the hint block                                                                             |
| `hint`     | `str \| list[TextBlock \| DataBlock]` | The hint payload — plain text or a list of multimodal blocks                                                    |
| `source`   | `str \| None`                         | Optional sender / origin tag (typically a small JSON object describing how the frontend should label this hint) |

**CustomEvent** — Generic extensible event used by service-layer middleware to notify subscribers about state changes (task progress, team membership, permission updates, …) without polluting the core agent event enum.

| Field      | Type   | Description                                                |
| ---------- | ------ | ---------------------------------------------------------- |
| `reply_id` | `str`  | ID of the reply message                                    |
| `name`     | `str`  | The signal name (e.g. `"tasks_context"`, `"team_updated"`) |
| `value`    | `dict` | Arbitrary JSON-serialisable payload for this signal        |

Reconstruct Messages from Events

Events and messages are not independent — they are two views of the same data. Every event produced by reply_stream can be applied to a Msg via append_event(), reconstructing the complete message incrementally. This guarantees that the final message state is fully recoverable from the event stream alone.

from agentscope.message import Msg, AssistantMsg

msg = None

# Accumulate events into the message
async for event in agent.reply_stream(user_msg):
	if isinstance(event, ReplyStartEvent):
		# Create a new message when the reply starts
		msg = AssistantMsg(name=event.name, content=[], id=event.reply_id)

	else:
		# For all other events, append to the message to reconstruct its state
        msg.append_event(event)

The append_event method handles all event types:

Event Type	Effect on Msg
ReplyEndEvent	Sets finished_at timestamp
TextBlockStartEvent	Appends a new empty TextBlock
TextBlockDeltaEvent	Concatenates delta to the block's text
DataBlockStartEvent	Appends a new empty DataBlock
DataBlockDeltaEvent	Concatenates data to the block's base64 content
ThinkingBlockStartEvent	Appends a new empty ThinkingBlock
ThinkingBlockDeltaEvent	Concatenates delta to the block's thinking text
ToolCallStartEvent	Appends a new ToolCallBlock with empty input
ToolCallDeltaEvent	Concatenates delta to the tool call's input
ToolResultStartEvent	Appends a new ToolResultBlock with empty output
ToolResultTextDeltaEvent	Appends text to the tool result's output
ToolResultDataDeltaEvent	Appends a binary data block to the tool result's output
ToolResultEndEvent	Sets the tool result's final state
HintBlockEvent	Appends a HintBlock to content (carrying the event's hint and source) so the hint is persisted and replayable
RequireUserConfirmEvent	Updates tool call states to ASKING
ExternalExecutionResultEvent	Appends ToolResultBlocks to content

This design enables flexible deployment: a backend can stream events over WebSocket to a frontend, which reconstructs the message client-side. If the connection drops, replaying the event sequence from any checkpoint restores the exact message state.

TypeScript Support

A TypeScript version of the message and event primitives is available on npm, so frontends can reconstruct messages from the event stream with the same appendEvent API:

pnpm install @agentscope-ai/agentscope

import { AssistantMsg, ReplyStartEvent } from "@agentscope-ai/agentscope/message";

let msg: AssistantMsg | null = null;

for await (const event of stream) {
    if (event.type === "REPLY_START") {
        msg = new AssistantMsg({ name: event.name, content: [], id: event.reply_id });
    } else {
        msg?.appendEvent(event);
    }
}

Example: Streaming UI

A typical pattern for building a streaming interface:

from agentscope.message import AssistantMsg, UserMsg
from agentscope.event import (
    ReplyStartEvent,
    TextBlockDeltaEvent,
    ToolCallStartEvent,
    ToolResultEndEvent,
    ReplyEndEvent,
)

msg = None

async for event in agent.reply_stream(UserMsg("user", "Fix the bug")):
    if isinstance(event, ReplyStartEvent):
        msg = AssistantMsg(name=event.name, content=[], id=event.reply_id)

    elif isinstance(event, TextBlockDeltaEvent):
        print(event.delta, end="", flush=True)

    elif isinstance(event, ToolCallStartEvent):
        print(f"\n[Calling {event.tool_call_name}...]")

    elif isinstance(event, ToolResultEndEvent):
        print(f"[Tool finished: {event.state}]")

    elif isinstance(event, ReplyEndEvent):
        print("\n[Done]")

    # Always accumulate into the message
    if msg is not None:
        msg.append_event(event)

# msg now contains the complete reply

Further Reading

How the agent produces events and messages in the ReAct loop How messages are stored, compressed, and offloaded

Tool

Define, register, and manage the capabilities an agent can call

Overview

Tools are how an agent acts on the world — running shell commands, reading files, calling APIs. Each tool exposes itself to the LLM as a JSON Schema, and the agent invokes it through a unified streaming interface.

AgentScope organizes tool-related building blocks under three concepts:

• Tool — any class that satisfies the ToolBase interface, including the built-ins shipped with AgentScope and the FunctionTool / MCPTool adapters that wrap plain functions or MCP-server tools.
• Toolkit — the container that registers tools, MCP clients, and skills, exposes their JSON schemas to the model, and dispatches each tool call to the right tool object.
• Tool Group — a named bundle of tools, MCP clients, and skills that can be activated or deactivated as a unit. The agent toggles groups at runtime via the built-in meta tool to keep its context focused.

from agentscope.tool import Toolkit, Bash, Read, Write, Edit

toolkit = Toolkit(
    tools=[Bash(), Read(), Write(), Edit()],
)

A Toolkit created with tools alone exposes those tools in the special "basic" group, which is always active. Adding mcps, skills_or_loaders, or extra tool_groups extends what the agent can reach — see the sections below.

Python Tool

A Python tool is any object satisfying the ToolBase interface. AgentScope ships built-in tools for common operations and exposes the same interface for developers to build custom tools.

ToolBase Interface

ToolBase is the abstract base class every tool satisfies. The tables below list its attributes and methods.

Attributes that describe the tool to the agent and the runtime:

Attribute	Type	Description
name	str	The tool name presented to the agent
description	str	Agent-oriented description of what the tool does
input_schema	dict	JSON Schema defining the tool's parameters
is_concurrency_safe	bool	Whether the tool is safe to call in parallel
is_read_only	bool	Whether the tool only reads data without side effects
is_external_tool	bool	If True, execution is delegated externally (see Define External Execution Tool)
is_state_injected	bool	If True, the agent state is injected via the _agent_state argument
is_mcp	bool	Whether the tool comes from an MCP server
mcp_name	str | None	The MCP server name when is_mcp is True

Methods that hook into execution and the permission system:

Method	Required	Description
check_permissions(tool_input, context)	Yes	Runtime permission check before execution; returns PermissionDecision
check_read_only(tool_input)	Optional	Per-invocation read-only check; returns bool. Defaults to self.is_read_only. Override when read-only-ness depends on the input (e.g. Bash — ls is read-only, rm is not). Used by the permission engine to decide auto-allow in EXPLORE / ACCEPT_EDITS.
match_rule(rule_content, tool_input)	Optional	Custom rule-matching logic for the permission system; returns bool
generate_suggestions(tool_input)	Optional	Generate suggested permission rules from a tool call; returns list[PermissionRule]
__call__(**kwargs)	Optional	The tool's execution logic; returns ToolChunk or AsyncGenerator[ToolChunk, None]. Not required for external execution tools.

Use Built-in Tools

AgentScope ships a set of ready-to-use tools covering common agent operations. Instantiate them and pass into Toolkit(tools=[...]):

Tool	Description	Read-only
Bash	Execute shell commands	No
Read	Read file contents with line numbers	Yes
Write	Create or overwrite files	No
Edit	Perform exact string replacements in files	No
Glob	Find files by glob pattern	Yes
Grep	Search file contents using ripgrep	Yes
TaskCreate	Create a structured task for progress tracking	No
TaskGet	Retrieve task details by ID	Yes
TaskList	List all tasks and their status	Yes
TaskUpdate	Update task status or metadata	No

Two more tools — the `reset_tools` meta tool and the `Skill` viewer — are auto-registered by `Toolkit` whenever extra tool groups or skills exist. Developers do not instantiate them directly. See [Manage Tools Agentically](#manage-tools-agentically) and [Skill](#skill).

Bash

The Bash tool executes shell commands and returns stdout/stderr. It implements every optional interface method to provide fine-grained permission control.

check_permissions() runs a layered safety analysis on the command string:

1. Injection risk detection — flags dynamic shell structures ($(...), backticks, process substitution) that cannot be statically analyzed → ASK
2. Read-only command detection — auto-allows safe commands (git status, ls, cat, grep, docker ps, etc.), including compound commands where every subcommand is read-only → ALLOW
3. Dangerous command patterns — detects destructive operations (e.g. chmod 777, mkfs) → ASK
4. Sed constraint check — blocks in-place sed -i against dangerous files → ASK
5. Dangerous path protection — checks if the command operates on sensitive config files (.bashrc, .ssh/, .env) → ASK
6. Dangerous removal detection — catches rm / rmdir targeting critical system paths (/, ~, /usr) → ASK
7. ACCEPT_EDITS mode — auto-allows filesystem commands (mkdir, touch, rm, rmdir, mv, cp, sed) only when every target path resolves inside a configured working directory. A command that touches any path outside the working set (e.g. cp /etc/hosts /tmp/x) falls through to PASSTHROUGH instead of auto-allowing.

check_read_only() returns True for any command identified by the read-only detector above (step 2), and False otherwise. The permission engine uses it to decide auto-allow in EXPLORE / ACCEPT_EDITS without re-running the full safety analysis.

match_rule() uses prefix-based wildcard matching against the command string:

Pattern	Matches	Does not match
npm run:*	npm run build, npm run test	npm install
git commit:*	git commit -m "fix"	git push
rm:*	rm file.txt, rm -rf /tmp/x	ls

generate_suggestions() extracts the command prefix (first two tokens) and proposes a prefix rule. For example, git commit -m "fix bug" produces the suggestion git commit:*.

The constructor accepts optional extra entries for the dangerous-path lists:

from agentscope.tool import Bash

bash = Bash(
    additional_dangerous_files=[".secrets"],
    additional_dangerous_directories=[".credentials"],
)

File Tools (Read, Write, Edit)

The file tools enforce a read-before-write rule: Write and Edit require the target file to have been read via Read first. This prevents blind overwrites and ensures the agent always operates on current content.

Tool	Operation	Key behavior
Read	Read file contents	Returns content with line numbers; supports offset/limit for large files; results cached in agent state
Write	Create or overwrite a file	Fails if the file exists but has not been read first
Edit	Replace exact strings in a file	Fails if old_string is not found or is not unique (unless replace_all=True); requires prior read

check_permissions() — Write and Edit share the same permission logic:

1. Dangerous path protection — operations on sensitive files (.bashrc, .env, .ssh/) return a bypass-immune ASK (bypass_immune=True), so allow rules cannot silently authorize them. The ASK is still skipped in BYPASS mode (which opts out of safety prompts by design) and converted to DENY in DONT_ASK mode. See the permission system docs for the full contract.
2. ACCEPT_EDITS mode — auto-allows operations on files within configured working directories
3. PASSTHROUGH — falls through to the permission engine for rule matching

Read is read-only and always returns PASSTHROUGH (the engine handles EXPLORE-mode and ACCEPT_EDITS-mode auto-allow via check_read_only).

match_rule() — all three tools use fnmatch glob matching against the file_path argument:

Pattern	Matches
src/**	Any file under src/
src/**/*.py	Python files under src/
config.json	Exact file match

generate_suggestions() proposes a glob covering the parent directory. For example, editing /project/src/main.py produces the suggestion src/**.

Create Custom Tool

To create a custom tool, subclass ToolBase, declare its schema, and implement check_permissions and __call__:

from agentscope.tool import ToolBase, ToolChunk
from agentscope.permission import (
    PermissionContext, PermissionDecision, PermissionBehavior,
)
from agentscope.message import TextBlock

class WebSearch(ToolBase):
    name = "WebSearch"
    description = "Search the web for information on a given query."
    input_schema = {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "The search query.",
            },
        },
        "required": ["query"],
    }
    is_concurrency_safe = True
    is_read_only = True

    async def check_permissions(
        self, tool_input: dict, context: PermissionContext,
    ) -> PermissionDecision:
        return PermissionDecision(
            behavior=PermissionBehavior.ALLOW,
            message="Web search is read-only.",
        )

    async def __call__(self, query: str) -> ToolChunk:
        results = await do_search(query)
        return ToolChunk(content=[TextBlock(text=results)])

Two extension hooks worth knowing about when writing custom tools with safety logic:

• check_read_only(tool_input) — override when whether an invocation modifies state depends on the input (like Bash: ls is read-only, rm is not). Defaults to returning the static is_read_only attribute. The permission engine calls it before deciding EXPLORE / ACCEPT_EDITS auto-allow.
• PermissionDecision(..., bypass_immune=True) — set on a returned ASK to mark it as a safety check that allow rules cannot silence (e.g. a DeployTool flagging prod-* targets). See the safety check contract for per-mode handling.

Wrap Function as Tool

For lightweight cases that don't justify a full subclass, wrap a plain Python function with the FunctionTool adapter. It auto-extracts the tool name from func.__name__, the description from the function docstring, and the input schema from type hints.

from agentscope.tool import FunctionTool, Toolkit

def get_weather(city: str, unit: str = "celsius") -> str:
    """Get the current weather for a city.

    Args:
        city: The city name to look up.
        unit: Temperature unit, either "celsius" or "fahrenheit".
    """
    return f"The weather in {city} is 22°{unit[0].upper()}"

toolkit = Toolkit(tools=[FunctionTool(get_weather)])

FunctionTool accepts overrides when the auto-extracted defaults are not what you want:

Argument	Type	Description
func	Callable	The Python function to wrap
name	str | None	Override the tool name (defaults to func.__name__)
description	str | None	Override the description (defaults to the docstring)
is_concurrency_safe	bool	Whether parallel calls are safe (default True)
is_read_only	bool	Whether the function has side effects (default False)
is_state_injected	bool	Whether the agent state is injected as _agent_state (default False)

Wrapped functions default to `ASK` permission behavior — the user must explicitly allow each call. Subclass `ToolBase` directly when you need custom permission logic.

Define External Execution Tool

An external execution tool delegates its actual execution outside the agent runtime — typically to a human operator or an external system. When the agent calls one, it emits a RequireExternalExecutionEvent and pauses until the result is delivered via ExternalExecutionResultEvent.

This pattern underlies the human-in-the-loop workflow, where certain actions require human approval or manual execution.

To create an external execution tool, set is_external_tool = True. There is no need to implement __call__:

from agentscope.tool import ToolBase
from agentscope.permission import (
    PermissionContext, PermissionDecision, PermissionBehavior,
)

class HumanApproval(ToolBase):
    name = "HumanApproval"
    description = "Request human approval for a sensitive operation."
    input_schema = {
        "type": "object",
        "properties": {
            "action": {"type": "string", "description": "The action requiring approval."},
            "reason": {"type": "string", "description": "Why this action needs approval."},
        },
        "required": ["action", "reason"],
    }
    is_concurrency_safe = True
    is_read_only = False
    is_external_tool = True

    async def check_permissions(
        self, tool_input: dict, context: PermissionContext,
    ) -> PermissionDecision:
        return PermissionDecision(
            behavior=PermissionBehavior.ALLOW,
            message="External tool dispatch is always allowed.",
        )

MCP

AgentScope integrates with Model Context Protocol (MCP) servers, letting an agent reach any MCP-compatible tool provider. The framework handles protocol negotiation, tool discovery, and result conversion automatically.

Two connection modes are supported:

• Stateful (STDIO or HTTP) — persistent session with explicit connect() / close() lifecycle
• Stateless (HTTP only) — ephemeral session created per tool call, no lifecycle management needed

MCP tools are namespaced as mcp__{server_name}__{tool_name} to avoid name collisions, and tools annotated with readOnlyHint are recognized as read-only by the permission system (auto-allowed in EXPLORE and ACCEPT_EDITS modes; in DEFAULT they still ASK unless an allow rule matches).

Register MCP Tool

Build one or more MCPClient instances and pass them to Toolkit(mcps=[...]). Stateful clients must be connected before the toolkit is constructed.

```python Stateful (STDIO) from agentscope.mcp import MCPClient, StdioMCPConfig from agentscope.tool import Toolkit

client = MCPClient( name="filesystem", is_stateful=True, mcp_config=StdioMCPConfig( command="mcp-server-filesystem", args=["--root", "/my/project"], ), )

await client.connect()

toolkit = Toolkit(mcps=[client])

```python Stateful (HTTP)
from agentscope.mcp import MCPClient, HttpMCPConfig
from agentscope.tool import Toolkit

client = MCPClient(
    name="weather",
    is_stateful=True,
    mcp_config=HttpMCPConfig(
        url="https://api.weather.com/mcp",
        headers={"Authorization": "Bearer xxx"},
    ),
)

await client.connect()

toolkit = Toolkit(mcps=[client])

from agentscope.mcp import MCPClient, HttpMCPConfig
from agentscope.tool import Toolkit

client = MCPClient(
    name="search",
    is_stateful=False,
    mcp_config=HttpMCPConfig(url="https://api.search.com/mcp"),
)

toolkit = Toolkit(mcps=[client])

To expose only a subset of an MCP server's tools, set enable_tools or disable_tools on the client itself:

client = MCPClient(
    name="search",
    is_stateful=False,
    mcp_config=HttpMCPConfig(url="https://api.search.com/mcp"),
    enable_tools=["web_search", "image_search"],
)

If you need to invoke MCP tools outside a Toolkit, call await client.list_tools() to retrieve a list of MCPTool adapters and use them like any other ToolBase instance.

Skill

Skills are markdown-based instruction sets that extend agent capabilities without writing new tool code. Each skill is a directory containing a SKILL.md file with frontmatter metadata and detailed instructions.

Unlike tools, skills are not callable directly. The agent uses the auto-registered Skill viewer tool to read a skill's instructions, then follows those instructions using its existing tools.

Register Skill

Pass skill sources to the Toolkit constructor through skills_or_loaders. Each entry can be a directory path string, a Skill object, or a SkillLoaderBase subclass:

```python Directory path (simple) from agentscope.tool import Toolkit

toolkit = Toolkit( skills_or_loaders=["/path/to/skills"], )

```python LocalSkillLoader (with subdirectory scanning)
from agentscope.tool import Toolkit
from agentscope.skill import LocalSkillLoader

loader = LocalSkillLoader(
    directory="/path/to/skills",
    scan_subdir=True,
)

toolkit = Toolkit(skills_or_loaders=[loader])

How Skill Works

When a Toolkit is constructed with skills, the registration and lookup flow runs in two phases.

At initialization:

• The toolkit scans every registered skill source and collects each skill's name, description, and directory.
• It auto-registers the built-in Skill viewer tool.
• It composes a system-prompt fragment listing the available skills (name and description only) and instructing the agent to invoke the Skill viewer to read the full content.

At runtime:

• The agent picks a skill by name and calls the Skill viewer.
• The viewer reads the corresponding SKILL.md and returns its full markdown.
• The agent follows those instructions using its already-equipped tools.

Skills are not tools — the agent cannot call a skill directly. It must first use the `Skill` viewer to read the instructions, then execute the steps described within using its other tools.

Manage Tools Agentically

The built-in meta tool (reset_tools) lets the agent self-manage which tool groups are active at runtime. This keeps its context focused — only tools relevant to the current task are exposed.

Define Tool Group

A ToolGroup is a named bundle of tools, MCP clients, and skills. Pass groups to Toolkit(tool_groups=[...]). The reserved "basic" group is created automatically from the constructor's top-level tools, mcps, and skills_or_loaders arguments and is always active.

from agentscope.tool import Toolkit, ToolGroup, Bash, Read, Write, Edit

toolkit = Toolkit(
    tools=[Bash(), Read(), Write(), Edit()],
    tool_groups=[
        ToolGroup(
            name="database",
            description="Tools for database operations.",
            instructions="Always wrap mutations in a transaction.",
            tools=[db_query_tool, db_migrate_tool],
        ),
        ToolGroup(
            name="deployment",
            description="Tools for deploying services.",
            instructions="Confirm the target environment before deploying.",
            tools=[deploy_tool, rollback_tool],
        ),
    ],
)

ToolGroup accepts the same tools, mcps, and skills_or_loaders arguments as the toolkit, plus a description shown to the agent in the meta tool schema and an optional instructions string returned when the group is activated.

Use Meta Tool

When at least one non-basic tool group exists, Toolkit auto-registers reset_tools and exposes its schema to the agent. Each non-basic group becomes a boolean field on that schema, and the agent calls the meta tool with the desired final state.

The behavior at runtime:

• Tools in the "basic" group are always exposed; they are never affected by the meta tool.
• Each call to reset_tools overwrites the activated set — any non-basic group not explicitly set to True becomes inactive, regardless of its previous state.
• For each group transitioning to active, its instructions (when provided) are concatenated and returned in the meta tool's response, telling the agent how to use that group properly.
• Tools from inactive groups are hidden from the agent's tool schema, freeing context space for the active set.

The meta tool input represents the **final state** of all groups, not incremental changes. Any group not explicitly set to `True` will be deactivated, regardless of its previous state.

Further Reading

How agents orchestrate tool calls in the ReAct loop Fine-grained control over which tools can execute and when Intercept and transform tool calls with onion-style middleware External execution tools and human approval workflows

Context

Manage agent context window for stable, long-running execution

Overview

The context is an agent's working memory — the messages (user inputs, assistant responses, tool calls, tool results) that the LLM sees on every reasoning step. As conversations grow, raw context eventually exceeds the model's window. AgentScope keeps an agent runnable indefinitely through three mechanisms:

• Context compression — summarizes older messages once token usage approaches the model's limit.
• Tool result truncation — caps oversized tool outputs before they enter the context.
• Context offloading — persists removed content to external storage so the agent can retrieve it later.

Before each model call, the agent assembles a single API input from three layers — the structure below shows what flows into that call:

  <Tree.File name="Skill instructions (from Toolkit)" />

  <Tree.File name="on_system_prompt middleware transforms" />
</Tree.Folder>

<Tree.File name="Summary (compressed history, if any)" />

<Tree.File name="Context (recent uncompressed messages)" />

</Tree.Folder>

How each layer is built:

1. System prompt — starts from the system_prompt passed at agent creation, then appends skill instructions (each skill's name and description, sourced from the toolkit), then runs every on_system_prompt middleware hook in order.
2. Summary — the compressed digest of older messages, present only after a compression has occurred.
3. Context — the recent uncompressed messages (user inputs, assistant responses, tool calls, tool results).

Use the `on_system_prompt` middleware hook to inject dynamic context — workspace instructions, time-sensitive facts, environment details — without rewriting the base prompt.

Compact Context

When the context window fills up, AgentScope keeps it in shape with two automatic mechanisms governed by ContextConfig: context compression (summarize older messages) and tool result truncation (cap oversized tool outputs). Both run transparently — the agent continues working without interruption.

Configure ContextConfig

ContextConfig is passed to the agent at construction time:

from agentscope.agent import Agent
from agentscope.agent import ContextConfig

agent = Agent(
    name="my_agent",
    system_prompt="...",
    model=model,
    toolkit=toolkit,
    context_config=ContextConfig(
        trigger_ratio=0.8,
        reserve_ratio=0.1,
        tool_result_limit=3000,
    ),
)

Available fields:

Parameter	Type	Description
trigger_ratio	float	Compression activates when token usage exceeds this ratio of the model's context size (capped at 0.9)
reserve_ratio	float	Proportion of context tokens kept as recent messages after compression
tool_result_limit	int	Maximum tokens per tool result; outputs exceeding this are truncated
compression_prompt	str	The prompt that guides the model to generate the summary
summary_template	str	String template for formatting the summary into the context
summary_schema	dict	JSON Schema constraining the model's structured summary output

Compress Context

Compression runs automatically before each reasoning step. The flow:

The agent totals the tokens of the system prompt, summary, context, and tool schemas. If total tokens exceed `trigger_ratio × context_size`, compression activates. Otherwise the agent proceeds with the model call as usual. Older messages are marked for compression; recent messages within `reserve_ratio × context_size` are kept. Tool call / result pairs are kept intact across the split. The model produces a structured summary from the older messages, with five fields: `task_overview`, `current_state`, `important_discoveries`, `next_steps`, `context_to_preserve`. The summary replaces the compressed messages; the reserved messages become the new context. The agent then continues its reasoning step. The remaining 10% between `trigger_ratio` (max `0.9`) and the full context size is reserved for the compression model call itself — the model needs room to generate the summary.

Compression can also be triggered manually by calling the agent's compress_context() method. Without arguments, it uses the agent's stored context_config; pass a one-off ContextConfig to override:

# Force-check using the agent's default config
await agent.compress_context()

# Or override the config for this single call (e.g. compress more aggressively)
from agentscope.agent import ContextConfig

await agent.compress_context(
    context_config=ContextConfig(trigger_ratio=0.5, reserve_ratio=0.1),
)

The method is a no-op when token usage is below trigger_ratio × context_size, so it is safe to call between turns or at any custom checkpoint.

Truncate Tool Results

After each tool call, the agent compares the result's token count against tool_result_limit. If the limit is exceeded, the result is split into a reserved portion (kept in context) and an offloaded portion (handed to the offloader if one is attached — see Offload Context).

A truncation marker is appended to the reserved portion so the agent knows the output was clipped:

<<<TRUNCATED>>>
<system-reminder>The remaining content has been omitted for limited context.</system-reminder>

When an offloader is attached, the marker also points the agent to the persisted full output:

<<<TRUNCATED>>>
<system-reminder>The remaining content has been omitted for limited context. You can refer to the file in '/path/to/tool_result-<id>.txt' for the truncated content if needed.</system-reminder>

Setting `tool_result_limit` too low may starve the agent of critical tool output. Setting it too high risks one result filling the entire context.

Offload Context

Context offloading writes content the agent has dropped — compressed messages, truncated tool outputs — to external storage, so the agent can read it back later via its file tools (Read, Grep, Glob) when it needs a detail that was compressed away.

Use the Offloader Protocol

Offloading is wired through the Offloader protocol — a structural contract with two methods:

Method	Description
offload_context(session_id, msgs)	Persist compressed messages; returns a reference (e.g. a file path) to the persisted content
offload_tool_result(session_id, tool_result)	Persist a truncated tool result; returns a reference to the persisted content

Pass any object satisfying this protocol to the agent's offloader argument. AgentScope's workspace module ships ready-made implementations:

from agentscope.agent import Agent
from agentscope.workspace import LocalWorkspace

workspace = LocalWorkspace(workdir="/tmp/agent_workspace")
await workspace.initialize()

agent = Agent(
    name="my_agent",
    system_prompt="...",
    model=model,
    toolkit=toolkit,
    offloader=workspace,
)

Without an offloader, compressed messages and truncated tool results are simply dropped after they leave the context window.

Use LocalWorkspace

LocalWorkspace writes offloaded content under workdir, isolating each agent run by session_id:

<Tree.Folder name="sessions" defaultOpen>
  <Tree.Folder name="{session_id}" defaultOpen>
    <Tree.File name="context.jsonl" />

    <Tree.File name="tool_result-{tool_id}.txt" />
  </Tree.Folder>
</Tree.Folder>

<Tree.Folder name="skills">
  <Tree.File name="..." />
</Tree.Folder>

</Tree.Folder>

How content is laid out:

• sessions/{session_id}/ — one directory per agent session, so concurrent agents don't collide. Compressed messages append to context.jsonl; each truncated tool result becomes its own tool_result-{tool_id}.txt.
• data/ — multimodal files (images, audio) referenced by offloaded messages, deduplicated by SHA-256 content hash.
• skills/ — unrelated to offloading; the workspace also serves as the agent's skill directory.

Create Custom Offloader

For backends other than the local filesystem (databases, cloud blobs, vector stores), implement the Offloader protocol — no inheritance required, since it is a structural protocol:

from typing import Any
from agentscope.message import Msg, ToolResultBlock

class S3Offloader:
    def __init__(self, bucket: str, prefix: str) -> None:
        self.bucket = bucket
        self.prefix = prefix

    async def offload_context(
        self,
        session_id: str,
        msgs: list[Msg],
        **kwargs: Any,
    ) -> str:
        key = f"{self.prefix}/sessions/{session_id}/context.jsonl"
        content = "\n".join(m.model_dump_json() for m in msgs)
        await self._upload(self.bucket, key, content)
        return f"s3://{self.bucket}/{key}"

    async def offload_tool_result(
        self,
        session_id: str,
        tool_result: ToolResultBlock,
        **kwargs: Any,
    ) -> str:
        key = f"{self.prefix}/sessions/{session_id}/tool_result-{tool_result.id}.txt"
        # Extract text content from the tool result blocks and upload.
        ...
        return f"s3://{self.bucket}/{key}"

Pass the instance into Agent(offloader=...) like any built-in workspace.

Further Reading

Built-in offloader implementations and the agent's working environment The ReAct loop and how context flows through reasoning steps Intercept model calls and system prompt composition with middleware hooks Tools that produce results subject to compression

Middleware

Intercept and extend agent behavior at key lifecycle points

Overview

Agent middleware is the mechanism for injecting custom logic — logging, tracing, input rewriting, access control — into key points of the agent execution pipeline, without modifying the agent or model code.

AgentScope exposes 6 hook positions plus a tool-provider hook, covering the full path from the outer reply process down to the raw model API call:

Position	Type	Description
on_reply	Onion	Wraps a complete reply, covering all ReAct rounds, tool executions, and the final output
on_reasoning	Onion	Wraps a single ReAct round's reasoning step (input assembly → model call → stream decoding)
on_acting	Onion	Wraps a single tool call execution
on_model_call	Onion	Wraps the underlying ChatModel API call — the closest to the model
on_compress_context	Onion	Wraps Agent.compress_context() — fires before each reasoning step when the agent decides whether to compress its context
on_system_prompt	Transformer	Fires every time the system prompt is assembled; multiple middlewares chain in sequence, each transforming the previous one's output
list_tools	Tool source	Optional. Returns a list[ToolBase] that the middleware contributes. Not invoked automatically — the caller assembling the agent's toolkit decides whether to call it and how to merge the result.

The three types differ as follows:

• Onion — middleware wraps the next handler, allowing logic before/after next_handler() and observation of the intermediate event stream.
• Transformer — middlewares form a pipeline; the previous one's output feeds into the next one. There is no "inner layer" concept.
• Tool source — not a hook on the runtime path. Agent.__init__ does not call list_tools(); you opt in explicitly by collecting the tools from your middlewares and passing them into the toolkit yourself.

The diagram below shows how these hooks nest within the agent lifecycle. on_system_prompt is embedded inside on_reasoning because it fires when the reasoning step assembles the system prompt; on_compress_context sits at the top of each ReAct round, before reasoning:

  <Tree.Folder name="on_reasoning" defaultOpen>
    <Tree.File name="on_system_prompt (system prompt assembly)" />

    <Tree.File name="on_model_call (model API call)" />
  </Tree.Folder>

  <Tree.Folder name="on_acting (once per tool call)" />
</Tree.Folder>

</Tree.Folder>

`on_acting` currently wraps only tool execution inside the agent runtime; tools dispatched outside the agent via external execution are not tracked by `on_acting`.

Equip Middleware

AgentScope packages a set of hooks into a class — a single middleware class can implement any subset of the 6 hook positions (plus the optional list_tools tool-provider hook) at the same time. Pass instances to Agent(middlewares=[...]) to equip them:

from agentscope.agent import Agent
from agentscope.middleware import TracingMiddleware

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
    middlewares=[TracingMiddleware()],
)

At construction time the agent scans each middleware instance, checks which hooks it actually implements, and routes it into the matching position-specific execution lists. Unimplemented positions are skipped automatically with no call overhead.

Built-in Middleware

TracingMiddleware

TracingMiddleware wires the full agent lifecycle to OpenTelemetry tracing. It instruments on_reply, on_model_call, and on_acting, producing hierarchical spans.

Before using it, register a TracerProvider and an OTLP exporter in the process:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")),
)
trace.set_tracer_provider(provider)

Then attach TracingMiddleware to the agent:

from agentscope.agent import Agent
from agentscope.middleware import TracingMiddleware

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
    middlewares=[TracingMiddleware()],
)

Each reply produces a nested span tree. The key attributes captured at each level are:

From `on_reply`:

* Agent name, session ID, reply ID
* Input messages and the final output message
* HITL pending tool calls
* External execution pending tool calls

From `on_model_call`:

* Model name, provider, input/output token counts
* Request and response message content
* Wraps streaming responses, writing attributes onto the final chunk

From `on_acting`:

* Tool name, call ID, input arguments
* Tool execution result

When no TracerProvider is configured, every hook short-circuits directly to next_handler() — no spans are created, no attributes are computed — making the overhead negligible.

When the agent receives an `ExternalExecutionResultEvent` (a tool executed outside the agent), `TracingMiddleware` synthesizes a compensating span for each external execution result, preserving full observability for tools run by external systems.

Add Additional Spans

To trace custom operations within the agent lifecycle, use the standard OpenTelemetry Python SDK directly. Obtain a tracer scoped to AgentScope and wrap any target code in a span:

from opentelemetry import trace
from agentscope import __version__

tracer = trace.get_tracer("agentscope", __version__)

with tracer.start_as_current_span(
    name="your_span_name",
    attributes={
        # Optional key-value pairs attached to the span,
        # e.g. function name, input arguments, or any custom metadata.
    },
    end_on_exit=True,
) as span:
    # your code here

These custom spans are emitted alongside AgentScope's built-in spans and delivered to the same OTLP collector configured in the TracerProvider.

Custom Middleware

Subclass MiddlewareBase and implement only the hooks you need — leave the rest alone.

The example below covers every position in a single middleware. Each onion hook receives an input_kwargs dict carrying the fields that flow into the wrapped layer; forward it with next_handler(**input_kwargs), or pass keyword arguments to override specific fields:

from typing import AsyncGenerator, Awaitable, Callable

from agentscope.agent import Agent
from agentscope.event import AgentEvent
from agentscope.message import Msg
from agentscope.middleware import MiddlewareBase
from agentscope.model import ChatResponse
from agentscope.tool import ToolBase


class FullObservabilityMiddleware(MiddlewareBase):
    """Observe every middleware position at once, plus contribute a tool."""

    async def on_reply(
        self,
        agent: Agent,
        # {"inputs": Msg | list[Msg] | UserConfirmResultEvent | ExternalExecutionResultEvent | None}
        input_kwargs: dict,
        next_handler: Callable[..., AsyncGenerator[AgentEvent | Msg, None]],
    ) -> AsyncGenerator[AgentEvent | Msg, None]:
        print(f"[reply] start for {agent.name}")
        async for item in next_handler(**input_kwargs):
            yield item
        print(f"[reply] end for {agent.name}")

    async def on_reasoning(
        self,
        agent: Agent,
        # {"tool_choice": ToolChoice | None}
        input_kwargs: dict,
        next_handler: Callable[..., AsyncGenerator[AgentEvent, None]],
    ) -> AsyncGenerator[AgentEvent, None]:
        print("[reasoning] start")
        async for event in next_handler(**input_kwargs):
            yield event
        print("[reasoning] end")

    async def on_model_call(
        self,
        agent: Agent,
        # {"messages": list[Msg], "tools": list[dict], "tool_choice": ToolChoice | None, "current_model": ChatModelBase}
        input_kwargs: dict,
        next_handler: Callable[
            ..., Awaitable[ChatResponse | AsyncGenerator[ChatResponse, None]]
        ],
    ) -> ChatResponse | AsyncGenerator[ChatResponse, None]:
        print(f"[model_call] {input_kwargs['current_model'].model}")
        result = await next_handler(**input_kwargs)
        print("[model_call] done")
        return result

    async def on_compress_context(
        self,
        agent: Agent,
        # {"context_config": ContextConfig | None}
        input_kwargs: dict,
        next_handler: Callable[..., Awaitable[None]],
    ) -> None:
        print(f"[compress_context] checking context for {agent.name}")
        await next_handler(**input_kwargs)
        print("[compress_context] done")

    async def on_system_prompt(
        self,
        agent: Agent,
        current_prompt: str,
    ) -> str:
        print(f"[system_prompt] length={len(current_prompt)}")
        return current_prompt

    async def list_tools(self) -> list[ToolBase]:
        # Optional hook. Not invoked automatically by ``Agent.__init__``;
        # if you want these tools available to the agent, collect them
        # from your middlewares yourself and pass them into the toolkit.
        return []

Execution Order

Onion hooks (on_reply, on_reasoning, on_acting, on_model_call) — the first middleware in the list is the outermost layer:

middlewares = [mw1, mw2]
# Call order:
# mw1 pre → mw2 pre → inner logic → mw2 post → mw1 post

For streaming / event-yielding hooks, the inner middleware sees each yielded event first:

mw1_pre → mw2_pre → mw2_event → mw1_event → ... → mw2_post → mw1_post

Transformer hooks (on_system_prompt) — middlewares chain left to right:

middlewares = [mw1, mw2]
# original_prompt → mw1.on_system_prompt() → mw2.on_system_prompt() → final

The overall execution order of all hooks within a single reply follows the agent lifecycle:

on_reply
  └── per ReAct round:
        ├── on_compress_context → compress_context()
        │     └── on_system_prompt (token counting before compression)
        ├── on_reasoning
        │     ├── _prepare_model_input() → on_system_prompt
        │     └── on_model_call
        └── on_acting (once per tool call in this round)

list_tools is not part of the per-reply execution path and is not invoked automatically by the agent — it is a convenience interface so a middleware can advertise its own tools. The caller assembling the toolkit decides whether to collect them.

Practical Examples

Timing middleware

The middleware below records the elapsed time of every model call:

import time
from agentscope.middleware import MiddlewareBase

class TimingMiddleware(MiddlewareBase):
    async def on_model_call(self, agent, input_kwargs, next_handler):
        model_name = input_kwargs["current_model"].model
        start = time.time()

        result = await next_handler()

        elapsed = time.time() - start
        print(f"[timing] {agent.name} → {model_name}: {elapsed:.2f}s")
        return result

Rate-limiting middleware

The middleware below enforces a minimum interval between two model calls:

import asyncio
import time
from agentscope.middleware import MiddlewareBase

class RateLimitMiddleware(MiddlewareBase):
    def __init__(self, min_interval: float = 1.0):
        self._last_call = 0.0
        self._min_interval = min_interval

    async def on_model_call(self, agent, input_kwargs, next_handler):
        now = time.time()
        wait = self._min_interval - (now - self._last_call)
        if wait > 0:
            await asyncio.sleep(wait)
        self._last_call = time.time()
        return await next_handler()

Dynamic system prompt middleware

The middleware below injects real-time context into the system prompt:

from datetime import datetime
from agentscope.middleware import MiddlewareBase

class DynamicContextMiddleware(MiddlewareBase):
    def __init__(self, context_fn):
        self._context_fn = context_fn

    async def on_system_prompt(self, agent, current_prompt):
        context = self._context_fn()
        return f"{current_prompt}\n\n## Current Context\n{context}"

agent = Agent(
    ...
    middlewares=[
        DynamicContextMiddleware(
            lambda: f"Time: {datetime.now().isoformat()}"
        ),
    ],
)

Model fallback middleware

The middleware below switches to a fallback model when the primary one fails:

from agentscope.middleware import MiddlewareBase

class ModelFallbackMiddleware(MiddlewareBase):
    def __init__(self, fallback_model):
        self._fallback = fallback_model

    async def on_model_call(self, agent, input_kwargs, next_handler):
        try:
            return await next_handler()
        except Exception as e:
            print(f"Primary model failed: {e}, switching to fallback")
            return await next_handler(
                current_model=self._fallback,
            )

Permission System

Fine-grained control over which tools your agents can execute and when

Overview

The permission system intercepts every tool call an agent makes and produces one of three decisions: allow the tool to execute, deny it, or ask the user for confirmation.

The system combines static configuration with dynamic runtime analysis. Three components drive the decision together:

• Rules — explicit allow/deny/ask patterns per tool and command, evaluated with highest priority. Rules have two sources: statically pre-configured in PermissionContext, or dynamically added when the user accepts a suggested rule during an ASK prompt. Suggestions are auto-generated from the current tool call, so accepting one means future identical calls are handled automatically without prompting again.
• Mode — a static global policy set at configuration time; determines default behavior for all calls that match no rules (e.g., EXPLORE makes the agent read-only; DONT_ASK silently denies unmatched calls).
• Built-in checks — dynamic runtime analysis performed by the tools themselves against actual call inputs: read-only command detection (parsing the bash command at call time) and dangerous path protection (checking the real file path or command target). Tools can mark a safety ASK as bypass-immune via PermissionDecision.bypass_immune=True; the engine then honors it across DEFAULT, ACCEPT_EDITS, and DONT_ASK (allow rules cannot silence it). BYPASS mode is the one exception — it explicitly opts out of all safety ASKs by design.

sequenceDiagram
    participant LLM
    participant PS as Permission System
    participant Tool
    participant User

    LLM->>PS: Tool Call
    Note over PS: Built-in Checks · Rules · Mode

    alt ALLOW
        PS->>Tool: execute
        Tool->>LLM: result
    else DENY
        PS->>LLM: denied
    else ASK + Suggestions
        PS->>User: ASK + Suggestions
        alt User approves
            User->>Tool: allow
            Tool->>LLM: result
            User-->>PS: accept suggested rule
        else User denies
            User->>PS: deny
            PS->>LLM: denied
        end
    end

Below is the decision flow for each mode. ASK outcomes trigger user confirmation; if the user accepts the auto-generated suggested rule, it is persisted for future calls.

```mermaid flowchart TD A([Tool Call]) --> D1{Deny Rules?} D1 -->|Match| DENY([DENY]) D1 -->|No| D2{Ask Rules?} D2 -->|Match| ASK([ASK]) D2 -->|No| D3[Tool check_permissions] D3 -->|ALLOW| ALLOW([ALLOW]) D3 -->|DENY| DENY D3 -->|"Safety ASK (bypass_immune)"| ASK D3 -->|PASSTHROUGH / other ASK| D4{Allow Rules?} D4 -->|Match| ALLOW D4 -->|No| ASK style DENY fill:#ff6b6b,color:#fff style ALLOW fill:#51cf66,color:#fff style ASK fill:#ffd43b,color:#333 ``` ```mermaid flowchart TD A([Tool Call]) --> E1{Deny Rules?} E1 -->|Match| DENY([DENY]) E1 -->|No| E2{Ask Rules?} E2 -->|Match| ASK([ASK]) E2 -->|No| E3{check_read_only?} E3 -->|True| ALLOW([ALLOW]) E3 -->|False| DENY style DENY fill:#ff6b6b,color:#fff style ALLOW fill:#51cf66,color:#fff style ASK fill:#ffd43b,color:#333 ``` ```mermaid flowchart TD A([Tool Call]) --> AE1{Deny Rules?} AE1 -->|Match| DENY([DENY]) AE1 -->|No| AE2{Ask Rules?} AE2 -->|Match| ASK([ASK]) AE2 -->|No| AE3{check_read_only?} AE3 -->|True| ALLOW([ALLOW]) AE3 -->|False| AE4[Tool check_permissions] AE4 -->|ALLOW| ALLOW AE4 -->|DENY| DENY AE4 -->|"Safety ASK (bypass_immune)"| ASK AE4 -->|PASSTHROUGH / other ASK| AE5{Allow Rules?} AE5 -->|Match| ALLOW AE5 -->|No| ASK style DENY fill:#ff6b6b,color:#fff style ALLOW fill:#51cf66,color:#fff style ASK fill:#ffd43b,color:#333 ``` ```mermaid flowchart TD A([Tool Call]) --> B1{Deny Rules?} B1 -->|Match| DENY([DENY]) B1 -->|No| B2{Ask Rules?} B2 -->|Match| ASK([ASK]) B2 -->|No| B3[Tool check_permissions] B3 -->|ALLOW| ALLOW([ALLOW]) B3 -->|DENY| DENY B3 -->|"Any ASK (safety ignored) / PASSTHROUGH"| B4{Allow Rules?} B4 -->|Match or No| ALLOW style DENY fill:#ff6b6b,color:#fff style ALLOW fill:#51cf66,color:#fff style ASK fill:#ffd43b,color:#333 ``` ```mermaid flowchart TD A([Tool Call]) --> DA1{Deny Rules?} DA1 -->|Match| DENY([DENY]) DA1 -->|No| DA2{Ask Rules?} DA2 -->|Match| DENY DA2 -->|No| DA3[Tool check_permissions] DA3 -->|ALLOW| ALLOW([ALLOW]) DA3 -->|DENY| DENY DA3 -->|"Any ASK (incl. safety)"| DENY DA3 -->|PASSTHROUGH| DA4{Allow Rules?} DA4 -->|Match| ALLOW DA4 -->|No| DENY style DENY fill:#ff6b6b,color:#fff style ALLOW fill:#51cf66,color:#fff ``` **Deny rules** and **explicit ask rules** are always honored, in every mode (including `BYPASS`).

Tool-emitted safety ASKs (bypass_immune=True) are honored in DEFAULT, ACCEPT_EDITS, and DONT_ASK — they cannot be silenced by allow rules. In BYPASS mode they are skipped on purpose: BYPASS's contract is "the user has opted out of safety prompts; only deny/ask rules remain as guardrails."

Permission Mode

AgentScope supports the following modes, each suited to a different deployment scenario.

Mode	Behavior	Use Case
DEFAULT	All ops require explicit rules or user confirmation. The only built-in auto-allow is Bash recognizing read-only commands (ls, git status, cat, ...) and returning ALLOW. Read / Glob / Grep return PASSTHROUGH and still ASK unless an allow rule matches	Most secure, recommended default
ACCEPT_EDITS	Auto-allow file ops in working directories; auto-allow Bash filesystem commands (mkdir/touch/rm/cp/mv/sed) only when every target path resolves inside a working directory	Active development with user present
EXPLORE	Read-only: allow read-only tools and read-only Bash commands (ls, git status, cat, ...); deny modifications. User-configured DENY/ASK rules take precedence over the read-only auto-allow	Code exploration, planning
BYPASS	Skip all permission checks except deny/ask rules and tool DENY. Tool safety ASKs are NOT enforced (rm -rf /, writes to ~/.bashrc, command injection, etc. all pass through). Use deny rules to protect specific paths	Sandboxed environments or fully trusted runs
DONT_ASK	Convert every ASK (default, ASK rules, and safety ASKs) to DENY; safe-by-default for non-interactive runs	Unattended / scheduled execution

Set the mode via AgentState.permission_context when creating the agent, or update it at runtime.

```python At initialization from agentscope.agent import Agent from agentscope.state import AgentState from agentscope.permission import PermissionContext, PermissionMode

agent = Agent( name="my_agent", system_prompt="...", model=model, state=AgentState( permission_context=PermissionContext( mode=PermissionMode.DEFAULT, ) ), )

```python At runtime
# Switch to read-only mode on the fly
agent.state.permission_context.mode = PermissionMode.EXPLORE

# Switch to unattended mode for batch execution
agent.state.permission_context.mode = PermissionMode.DONT_ASK

from agentscope.permission import AdditionalWorkingDirectory

agent = Agent(
    name="my_agent",
    system_prompt="...",
    model=model,
    state=AgentState(
        permission_context=PermissionContext(
            mode=PermissionMode.ACCEPT_EDITS,
            working_directories={
                "/my/project": AdditionalWorkingDirectory(
                    path="/my/project",
                    source="userSettings",
                )
            },
        )
    ),
)

Permission Rules

A PermissionRule maps a specific tool and call pattern to one of three behaviors: ALLOW, DENY, or ASK.

Each rule consists of the following fields. When the permission engine evaluates a rule, it calls the tool's match_rule() method with rule_content and the actual call input to determine whether the rule applies.

Tool this rule applies to: `"Bash"`, `"Read"`, `"Write"`, `"Edit"`, or any custom tool name. Match pattern — semantics depend on `tool_name`:

• Bash: wildcard prefix pattern (npm run:* matches npm run build, npm run test)
• Read / Write / Edit: glob pattern (src/**/*.py matches any .py under src/)
• Other tools: exact JSON-serialized parameter match

`ALLOW`, `DENY`, or `ASK` Origin of the rule: `"userSettings"`, `"projectSettings"`, `"session"`, etc.

Pattern Examples

rule_content is consumed by each tool's match_rule() method and auto-generated by ToolBase.generate_suggestions(). Because both methods are part of the tool interface, each tool can define its own pattern syntax and matching logic independently.

For AgentScope's built-in tools, the patterns are as follows:

Matches against the **`command`** parameter. Pattern format is `COMMAND_PREFIX:*` — the prefix is the leading token of the command, and `*` matches any arguments that follow.

| Pattern        | Matches                         | Does Not Match |
| -------------- | ------------------------------- | -------------- |
| `npm run:*`    | `npm run build`, `npm run test` | `npm install`  |
| `git commit:*` | `git commit -m "fix"`           | `git push`     |
| `rm:*`         | `rm file.txt`, `rm -rf /tmp/x`  | `ls`           |

```python
PermissionRule(
    tool_name="Bash",
    rule_content="npm run:*",
    behavior=PermissionBehavior.ALLOW,
    source="userSettings",
)
```

Matches against the **`file_path`** parameter using a glob pattern via `fnmatch`.

| Pattern       | Matches                   |
| ------------- | ------------------------- |
| `src/**`      | Any file under `src/`     |
| `src/**/*.py` | Python files under `src/` |
| `config.json` | Exact file match          |

```python
PermissionRule(
    tool_name="Write",
    rule_content="src/**",
    behavior=PermissionBehavior.ALLOW,
    source="userSettings",
)
```

Configuring Rules

At initialization — pass rules into PermissionContext when creating the agent:

from agentscope.agent import Agent
from agentscope.state import AgentState
from agentscope.permission import (
    PermissionContext, PermissionMode, PermissionRule, PermissionBehavior
)

agent = Agent(
    name="my_agent",
    system_prompt="...",
    model=model,
    state=AgentState(
        permission_context=PermissionContext(
            mode=PermissionMode.DEFAULT,
            allow_rules={
                "Bash": [PermissionRule(tool_name="Bash", rule_content="npm run:*",
                                        behavior=PermissionBehavior.ALLOW, source="userSettings")],
                "Write": [PermissionRule(tool_name="Write", rule_content="src/**",
                                         behavior=PermissionBehavior.ALLOW, source="userSettings")],
            },
            deny_rules={
                "Bash": [PermissionRule(tool_name="Bash", rule_content="rm:*",
                                        behavior=PermissionBehavior.DENY, source="userSettings")],
            },
        )
    ),
)

At runtime via suggestions — when the permission system returns ASK, it auto-generates suggested rules from the current call. Pass accepted rules back in UserConfirmResultEvent.rules; the agent adds them to the engine automatically:

from agentscope.event import UserConfirmResultEvent

# The ASK decision includes suggested_rules generated from the current call.
# To accept a suggestion, include it in the result event:
result = UserConfirmResultEvent(
    confirmed=True,
    rules=[suggested_rule],  # accepted rules are persisted to the engine
)

Built-in Checks

Each tool implements a check_permissions() method that runs against the actual call inputs at runtime. AgentScope's built-in tools cover three areas:

• Dangerous path protection — Write, Edit, and Bash check whether the target file or command touches sensitive paths. Returns a bypass-immune safety ASK that is honored in DEFAULT/ACCEPT_EDITS/DONT_ASK (allow rules cannot silence it). BYPASS mode skips it on purpose.
• Read-only command detection — Bash parses the command string to detect read-only operations and auto-allows them in every mode (including DEFAULT). For input-dependent tools like Bash, this is exposed via the check_read_only() method (see below).
• ACCEPT_EDITS mode — Write and Edit auto-allow operations on files within configured working directories. Bash additionally requires that every target path of a filesystem command (mkdir/touch/rm/cp/mv/sed, ...) resolves inside a working directory.

Custom tools

A custom tool implements check_permissions() to add tool-specific permission logic. Tools whose read-only status depends on the input (like Bash — ls is read-only, rm is not) should also override check_read_only().

from agentscope.tool import ToolBase
from agentscope.permission import PermissionContext, PermissionDecision, PermissionBehavior

class MyTool(ToolBase):
    name = "MyTool"
    # Static default. For tools whose answer depends on input, leave
    # this as the conservative default and override check_read_only().
    is_read_only = False

    async def check_read_only(self, tool_input: dict) -> bool:
        """Optional: dynamic read-only check.

        Defaults to returning self.is_read_only. Override when whether
        an invocation modifies state depends on the input. The engine
        calls this to decide auto-allow in EXPLORE and ACCEPT_EDITS.
        """
        return tool_input.get("operation") in {"list", "describe", "get"}

    async def check_permissions(
        self,
        tool_input: dict,
        context: PermissionContext,
    ) -> PermissionDecision:
        target = tool_input.get("target")

        # Custom safety check: block operations on production resources.
        # Setting bypass_immune=True makes this ASK survive allow rules
        # in DEFAULT/ACCEPT_EDITS/DONT_ASK; BYPASS still skips it.
        if target and target.startswith("prod-"):
            return PermissionDecision(
                behavior=PermissionBehavior.ASK,
                message=f"Operation targets production resource: {target}",
                decision_reason="Safety check: production resource",
                bypass_immune=True,
            )

        # Return PASSTHROUGH to let the engine continue with rules/mode
        return PermissionDecision(behavior=PermissionBehavior.PASSTHROUGH)

Safety check contract

A safety check is a tool-emitted ASK that the tool considers too dangerous to be silently overridden — e.g. Write to ~/.bashrc, Bash with rm -rf /. Setting bypass_immune=True on the decision asks the engine to surface the ASK to the user even when an allow rule matches or the mode would otherwise auto-allow.

Use it whenever a wrong call would cause damage the user almost certainly didn't intend. Example: a custom DeployTool returns bypass_immune=True when the target is prod-*, so a blanket allow_rules["DeployTool"] = ["*"] configured for staging cannot accidentally authorize a production deploy.

The exact handling per mode:

Mode	bypass_immune=True ASK is...
DEFAULT	honored — allow rules cannot override it
ACCEPT_EDITS	honored — same as DEFAULT
EXPLORE	not applicable (the engine does not call check_permissions in EXPLORE; the read-only verdict is final)
BYPASS	ignored — BYPASS skips all safety ASKs by design
DONT_ASK	converted to DENY (no user available to answer)

A regular ASK (bypass_immune=False, the default) can be overridden by a matching allow rule in DEFAULT/ACCEPT_EDITS, and is silently allowed by BYPASS's fallback.

Read-Only Commands

Common read-only bash commands are auto-allowed without any rules, in every mode (including DEFAULT). A compound command (&&, ||, ;, |) is read-only only if all subcommands are read-only. Output redirections (>, >>) always make a command non-read-only.

| Category | Commands | | ---------------- | ------------------------------------------------------------------------------------------------------------------------- | | Git | `git status`, `git log`, `git diff`, `git show`, `git branch`, `git blame`, `git grep`, `git reflog`, `git config --list` | | Files | `ls`, `cat`, `head`, `tail`, `grep`, `rg`, `find`, `tree`, `stat`, `wc`, `pwd`, `which` | | Docker | `docker ps`, `docker images`, `docker logs`, `docker inspect`, `docker info` | | GitHub CLI | `gh repo view`, `gh issue list`, `gh pr list`, `gh status` | | Package managers | `npm list`, `pip list`, `pip show`, `node --version`, `python --version` |

Dangerous Path Protection

Operations targeting the following paths trigger a bypass-immune ASK in `DEFAULT`, `ACCEPT_EDITS`, and `DONT_ASK` (converted to DENY in `DONT_ASK`). `BYPASS` mode explicitly skips this check — if you need dangerous-path protection while running in BYPASS, add deny rules for the specific paths.

Category	Paths
Shell configs	.bashrc, .zshrc, .bash_profile, .profile
Git configs	.gitconfig, .gitmodules
SSH	.ssh/config, .ssh/authorized_keys, id_rsa, id_ed25519
Credentials	.env, .env.local, .npmrc, .pypirc, .aws/credentials
Directories	.git/, .ssh/, .claude/, .vscode/, .aws/, .kube/

Common Recipes

The following examples show how to configure AgentState.permission_context for common deployment scenarios. Each recipe combines a mode with rules to match a specific use case.

```python Read-only exploration # EXPLORE mode: agent can freely use read-only tools (Read, Grep, # Glob) and read-only bash commands (`ls`, `git status`, `cat`, ...). # Any modification — Write, Edit, or non-read-only bash command — is # denied automatically. agent = Agent( name="explorer", system_prompt="...", model=model, state=AgentState( permission_context=PermissionContext(mode=PermissionMode.EXPLORE) ), ) ```

from agentscope.permission import PermissionRule, PermissionBehavior

agent = Agent(
    name="ci_agent",
    system_prompt="...",
    model=model,
    state=AgentState(
        permission_context=PermissionContext(
            mode=PermissionMode.DONT_ASK,
            allow_rules={
                "Bash": [
                    PermissionRule(tool_name="Bash", rule_content="npm run:*",
                                   behavior=PermissionBehavior.ALLOW, source="project"),
                    PermissionRule(tool_name="Bash", rule_content="git commit:*",
                                   behavior=PermissionBehavior.ALLOW, source="project"),
                ],
            },
        )
    ),
)
# Only explicitly allowed commands run; everything else (including
# safety ASKs like `rm -rf /` or writes to ~/.bashrc) is converted
# to DENY. Prefer DONT_ASK over BYPASS for unattended runs — it keeps
# the tools' safety net while still never prompting the user.

# BYPASS skips tools' safety ASKs by design — deny rules become the
# only guardrail. Always pair BYPASS with deny rules for the paths
# and commands you want to protect.
agent = Agent(
    name="my_agent",
    system_prompt="...",
    model=model,
    state=AgentState(
        permission_context=PermissionContext(
            mode=PermissionMode.BYPASS,
            deny_rules={
                "Bash": [
                    PermissionRule(tool_name="Bash", rule_content="rm:*",
                                   behavior=PermissionBehavior.DENY, source="userSettings"),
                    PermissionRule(tool_name="Bash", rule_content="git push:*",
                                   behavior=PermissionBehavior.DENY, source="userSettings"),
                ],
                "Write": [
                    PermissionRule(tool_name="Write", rule_content="**/.bashrc",
                                   behavior=PermissionBehavior.DENY, source="userSettings"),
                    PermissionRule(tool_name="Write", rule_content="**/.ssh/**",
                                   behavior=PermissionBehavior.DENY, source="userSettings"),
                ],
            },
        )
    ),
)
# Everything allowed except the deny-listed commands and paths.
# Without these deny rules, BYPASS would let the agent rm anything,
# git push anywhere, or overwrite ~/.bashrc — by design.

Workspace

The execution environment that supplies tools, skills, and context offloading

Overview

A workspace is the agent's execution environment. It supplies the agent with three categories of resources — tools (built-in tools and MCPs), skills, and context offloading for compressed messages and oversized tool results — and owns the lifecycle of the resources living inside it (MCP server processes, dynamically added skills, offloaded files).

AgentScope ships three workspace implementations — local filesystem, Docker container, and E2B cloud sandbox — plus a workspace manager that allocates and tracks workspaces in Agent Service so that multi-tenant deployments can map workspaces to users, agents, or sessions without rewriting the agent code.

For Docker and E2B, MCP servers run inside the isolated environment; the host reaches them through an in-workspace gateway covered in MCP Gateway.

Use a Workspace

Create a Workspace

AgentScope ships three workspace implementations, one per execution backend:

Class	Backend
LocalWorkspace	Host filesystem (built-in tools run host-side)
DockerWorkspace	Docker container via aiodocker
E2BWorkspace	E2B cloud sandbox via AsyncSandbox

Each backend has its own persistence model and directory layout — pick the tab matching your target environment:

`LocalWorkspace` persists state directly under `workdir` on the host filesystem — restarts simply re-open the same directory. The directory has the following layout:

```
{workdir}/
├── .mcp           # registered MCP client configs (JSON array)
├── data/          # offloaded multimodal payloads (deduped by SHA-256)
├── skills/        # skill subdirectories, each with SKILL.md
│   └── .skills    # name/hash index for de-duplication
└── sessions/      # per-session context.jsonl and tool-result files
```

```python
from agentscope.workspace import LocalWorkspace

workspace = LocalWorkspace(
    workdir="/data/my-workspace",
    default_mcps=[],
    skill_paths=["./skills/web-search"],
)
await workspace.initialize()
```

`DockerWorkspace` bind-mounts the host `workdir` to `/workspace` inside the container, so the layout below lives on the host and survives container restarts. Omit `workdir` for a purely ephemeral container whose writable layer disappears with it.

```
{workdir}/         # host directory, bind-mounted to /workspace in container
├── .mcp           # registered MCP client configs (JSON array)
├── data/          # offloaded multimodal payloads
├── skills/        # skill subdirectories, each with SKILL.md
└── sessions/      # per-session context.jsonl and tool-result files
```

```python
from agentscope.workspace import DockerWorkspace

workspace = DockerWorkspace(
    base_image="python:3.11-slim",
    workdir="/data/docker-workspaces/agent-1",  # bind-mounted to /workspace
    node_version="20",
    extra_pip=["numpy", "pandas"],
    default_mcps=[],
    skill_paths=["./skills/web-search"],
)
await workspace.initialize()
```

`E2BWorkspace` uses the sandbox filesystem itself as the persistence layer — there is no host `workdir`. Each sandbox is tagged with `workspace_id` in its E2B metadata; on restart the manager looks it up via `AsyncSandbox.list(...)` and reconnects with `connect(sandbox_id=...)`. Pausing keeps disk state, resuming restores it intact.

```
$workdir/          # inside the sandbox
├── .mcp           # registered MCP client configs (JSON array)
├── data/          # offloaded multimodal payloads
├── skills/        # skill subdirectories, each with SKILL.md
└── sessions/      # per-session context.jsonl and tool-result files
```

```python
from agentscope.workspace import E2BWorkspace

workspace = E2BWorkspace(
    template="base",
    api_key="your-e2b-api-key",   # or set E2B_API_KEY
    timeout_seconds=300,
    default_mcps=[],
    skill_paths=["./skills/web-search"],
)
await workspace.initialize()
```

default_mcps and skill_paths are seed-time inputs — they populate a brand-new workspace on first initialize(). On restart the workspace restores its MCP list from the persisted .mcp file, so re-passing the seeds is a no-op.

All three implementations share the same WorkspaceBase interface, so the same agent code runs against any backend. The methods fall into four roles:

Provision the workspace, release its resources, and clear per-session state for pool reuse. Also drives the `async with` protocol. Enumerate the built-in tools, active `MCPClient` instances, available skills, and a workspace-specific system prompt fragment. Persist compressed context or oversized tool results into `sessions//` and return a reference path the agent stores in place of the original payload. Register or remove MCP servers and skills at runtime; changes are persisted to `.mcp` and `skills/` so they survive restarts.

Integrate with Agent

A workspace plugs into Agent along two axes — as a source of tools / MCPs / skills, and as the offloader for context compression:

from agentscope.agent import Agent
from agentscope.tool import Toolkit
from agentscope.workspace import LocalWorkspace

workspace = LocalWorkspace(workdir="./my-workspace")
await workspace.initialize()

agent = Agent(
    name="coder",
    system_prompt="You are a coding assistant.",
    model=model,
    toolkit=Toolkit(
        tools=await workspace.list_tools(),
        mcps=await workspace.list_mcps(),
        skills_or_loaders=await workspace.list_skills(),
    ),
    offloader=workspace,
)

Axis	Wiring	What the agent gets
Resources	Toolkit(tools=..., mcps=..., skills_or_loaders=...)	Built-in tools, MCP-provided tools, and skills available in the workspace
Offloading	Agent(offloader=workspace)	When context compression triggers or a tool result exceeds the size limit, the agent calls workspace.offload_context() / offload_tool_result() and stores the returned reference path in place of the original payload

Agent only depends on the Offloader protocol (offload_context / offload_tool_result), so any object satisfying that protocol can take the role — the workspace is the typical implementation.

Because the workspace exposes its resources as flat lists, you can partition them into `ToolGroup`s when the agent has too many tools to keep all active at once. Pass the groups to `Toolkit(tool_groups=[...])` and the agent activates them on demand through the built-in `ResetTools` meta-tool — only the reserved `basic` group stays always-on.

from agentscope.tool import Toolkit, ToolGroup

mcps = await workspace.list_mcps()
skills = await workspace.list_skills()

toolkit = Toolkit(
    tools=await workspace.list_tools(),         # always active (basic group)
    tool_groups=[
        ToolGroup(name="search", description="Web search and retrieval.",
                  mcps=[m for m in mcps if m.name.startswith("search")]),
        ToolGroup(name="coding", description="Code editing skills.",
                  skills_or_loaders=skills),
    ],
)

Workspace Manager

A workspace manager is the allocator and lifecycle owner for workspaces in a multi-tenant service. It is used by Agent Service to map incoming requests to the right workspace instance and to release them on shutdown.

A manager is responsible for:

• Allocation — create_workspace(user_id, agent_id, session_id) builds a fresh workspace and tracks it; get_workspace(..., workspace_id) returns a live one or rebuilds on cache miss.
• Caching and TTL — workspaces are cached in memory keyed by workspace_id; idle entries are evicted after ttl seconds and their underlying resources (containers, sandboxes, MCP processes) torn down.
• Isolation policy — the manager decides whether two requests share a workspace; the built-in managers all isolate by agent_id, but WorkspaceManagerBase is a tiny abstract class that can be subclassed for per-user or per-session policies.
• Cleanup — close(workspace_id) evicts a single entry and close_all() drains the cache on app shutdown.

AgentScope ships three managers, each paired with one workspace backend and all isolating per agent:

Class	Pairs with	Isolation key	Cache key
LocalWorkspaceManager	LocalWorkspace	agent_id (workdir = <basedir>/<agent_id>)	workspace_id
DockerWorkspaceManager	DockerWorkspace	(user_id, agent_id) (workdir = <basedir>/<user_id>/<agent_id>)	workspace_id
E2BWorkspaceManager	E2BWorkspace	agent_id (workspace metadata, no host workdir)	workspace_id

Construct a manager and use it like a workspace allocator:

from agentscope.app._manager import LocalWorkspaceManager

manager = LocalWorkspaceManager(
    basedir="/data/workspaces",
    skill_paths=["./skills/coding"],
    ttl=3600.0,
)

ws = await manager.create_workspace(
    user_id="user-1",
    agent_id="agent-42",
    session_id="session-abc",
)

# Later, on a follow-up request for the same workspace:
ws = await manager.get_workspace(
    user_id="user-1",
    agent_id="agent-42",
    session_id="session-abc",
    workspace_id=ws.workspace_id,
)

To plug a different isolation policy (per-user, per-session, or hybrid), subclass WorkspaceManagerBase and override get_workspace / create_workspace with your own keying — see Agent Service · Workspace implementation and isolation for how the service wires a manager into the request lifecycle.

In the agent service, the workspace manager is bound to the FastAPI app state during the lifespan and shared across all requests; routers acquire workspaces through `get_workspace_manager` dependency injection. See [Agent Service](/v2/deploy/agent-service) for the full integration.

MCP Gateway

DockerWorkspace and E2BWorkspace cannot register host-side MCP clients directly — the MCP servers live inside the container or sandbox, and stdio sessions cannot cross that boundary. AgentScope solves this with an MCP gateway: a lightweight FastAPI process that runs inside the workspace, owns the upstream MCP sessions, and exposes them over a single authenticated HTTP endpoint that the host talks to.

flowchart LR
    subgraph Host
        Agent --> Toolkit
        Toolkit --> GC["GatewayMCPClient<br/>(MCPClient subclass)"]
    end

    subgraph Sandbox["Container / E2B Sandbox"]
        GC -- "HTTPS<br/>Bearer token" --> GW["MCP Gateway<br/>(FastAPI)"]
        GW --> MCP1["MCP Server 1 (stdio)"]
        GW --> MCP2["MCP Server 2 (http)"]
        GW --> MCPN["MCP Server N"]
    end

The gateway exposes a small REST surface — GET /health, GET/POST/DELETE /mcps, GET /mcps/{name}/tools, POST /mcps/{name}/tools/{tool} — protected by a per-workspace bearer token minted at each initialize(). On the host, two adapters preserve the standard interfaces:

• GatewayMCPClient — an MCPClient subclass whose connect / close / list_tools calls become HTTP requests against the gateway, so the rest of the toolkit cannot tell it apart from a local MCP client.
• GatewayMCPTool — a ToolBase subclass whose __call__ posts to /mcps/{name}/tools/{tool} and reconstructs the returned ToolChunk.

This abstraction keeps the agent-side code identical across all three workspace backends — a workspace returns MCPClient instances from list_mcps() regardless of whether the upstream session lives on the host (LocalWorkspace) or inside an isolated environment (DockerWorkspace / E2BWorkspace).

Further Reading

Agent abstraction, the ReAct loop, and offloader integration Built-in tools, MCP integration, and toolkit composition Multi-tenant service that drives the workspace manager Context compression and the offloading triggers