Agentscope Developer Agent Skill

┌─────────────────────────────────────────────────────────────┐
│                    AgentScope Ecosystem                     │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐   │
│  │   ReAct    │  │   Voice     │  │  Multi-Agent    │   │
│  │   Agent    │  │   Agent     │  │   Workflows     │   │
│  └─────────────┘  └─────────────┘  └─────────────────┘   │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐   │
│  │   Memory    │  │    Tools    │  │  Model Tuner    │   │
│  │ (InMem/ReMe)│  │ (MCP/A2A)  │  │   (RL/Finetune) │   │
│  └─────────────┘  └─────────────┘  └─────────────────┘   │
├─────────────────────────────────────────────────────────────┤
│  Deployment: Local | Serverless | K8s | Docker             │
└─────────────────────────────────────────────────────────────┘

# Installation
pip install agentscope
# Or with uv
uv pip install agentscope

# From source
git clone -b main https://github.com/agentscope-ai/agentscope.git
cd agentscope
pip install -e .

Build a simple ReAct agent named "Friday" that can execute Python code

from agentscope.agent import ReActAgent, UserAgent
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.tool import Toolkit, execute_python_code
import os, asyncio

async def main():
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)

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

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

asyncio.run(main())

Create a multi-agent discussion with 3 agents: a scientist, engineer, and artist

from agentscope.agent import ReActAgent
from agentscope.pipeline import MsgHub, sequential_pipeline
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
import asyncio

async def multi_agent_discussion():
    # Create agents
    scientist = ReActAgent(
        name="Scientist",
        sys_prompt="You are a scientist specializing in AI research.",
        model=DashScopeChatModel(model_name="qwen-max"),
    )
    engineer = ReActAgent(
        name="Engineer",
        sys_prompt="You are an engineer focused on practical implementations.",
        model=DashScopeChatModel(model_name="qwen-max"),
    )
    artist = ReActAgent(
        name="Artist",
        sys_prompt="You are an artist interested in creative AI applications.",
        model=DashScopeChatModel(model_name="qwen-max"),
    )

    # Create message hub
    async with MsgHub(
        participants=[scientist, engineer, artist],
        announcement=Msg("Host", "Discuss the future of AI.", "assistant")
    ) as hub:
        # Sequential discussion
        await sequential_pipeline([scientist, engineer, artist])

asyncio.run(multi_agent_discussion())

Build a voice-enabled agent that can speak responses

from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope.tools import TextToSpeechTool
from agentscope.memory import InMemoryMemory
import asyncio

async def voice_agent():
    agent = ReActAgent(
        name="VoiceAssistant",
        sys_prompt="You are a helpful voice assistant.",
        model=DashScopeChatModel(model_name="qwen-max"),
        memory=InMemoryMemory(),
        tools=[TextToSpeechTool()],
    )

    # Process voice input and respond with speech
    while True:
        user_input = await get_voice_input()  # Your STT implementation
        response = await agent(user_input)
        await play_speech(response)  # TTS output

asyncio.run(voice_agent())

Create an agent that remembers user preferences across sessions

from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope.memory import ReMeMemory
import asyncio

async def remember_user():
    # Initialize ReMe (Retrieve, Memorize, Evolve)
    memory = ReMeMemory(
        embedding_model="text-embedding-v3",
        recall_threshold=0.7,
        compression_threshold=50,
    )

    agent = ReActAgent(
        name="PersonalAssistant",
        sys_prompt="Remember user preferences and provide personalized help.",
        model=DashScopeChatModel(model_name="qwen-max"),
        memory=memory,
    )

    # First conversation
    await agent("I prefer concise answers")
    # Second conversation (memory persists)
    response = await agent("How's the weather?")
    # Agent recalls: user prefers concise answers

asyncio.run(remember_user())

Integrate external MCP tools like a maps API into my agent

from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope.mcp import HttpStatelessClient
from agentscope.tool import Toolkit
import os, asyncio

async def mcp_agent():
    # Initialize MCP client
    client = HttpStatelessClient(
        name="maps_mcp",
        transport="streamable_http",
        url=f"https://mcp.amap.com/mcp?key={os.environ['GAODE_API_KEY']}",
    )

    # Get tool as local callable function
    maps_geo = await client.get_callable_function(func_name="maps_geo")

    # Register in toolkit
    toolkit = Toolkit()
    toolkit.register_tool_function(maps_geo)

    agent = ReActAgent(
        name="TravelAssistant",
        sys_prompt="You are a travel assistant that can find locations.",
        model=DashScopeChatModel(model_name="qwen-max"),
        toolkit=toolkit,
    )

    # Use tool
    result = await agent("Find Tiananmen Square in Beijing")

asyncio.run(mcp_agent())

# Retry with Budget overrun
from agentscope.tools import retry_with_backoff

@retry_with_backoff(max_retries=3, initial_delay=1.0)
async def call_model_with_retry(agent, msg):
    return await agent(msg)

# Vendor non-performance for tools
from agentscope.tools import CircuitBreaker

breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60)

# Compliance violation
try:
    result = await agent(msg)
except Exception as e:
    result = "I'm having trouble processing your request. Please try again."