3. Quick Start¶

This tutorial walks through building a simple agent application in AgentScope Runtime and deploying it as a service.

Prerequisites¶

🔧 Installation Requirements¶

Install AgentScope Runtime with the base dependencies:

pip install agentscope-runtime

🔑 API Key Configuration¶

Provide an API key for your selected LLM provider. This example uses DashScope’s Qwen model, so we set the DashScope API key as an environment variable:

export DASHSCOPE_API_KEY="your_api_key_here"

Step-by-Step Implementation¶

Step 1: Import Dependencies¶

Start by importing every required module:

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeChatFormatter
from agentscope.tool import Toolkit, execute_python_code
from agentscope.pipeline import stream_printing_messages
from agentscope.memory import InMemoryMemory
from agentscope.session import RedisSession

from agentscope_runtime.engine import AgentApp
from agentscope_runtime.engine.schemas.agent_schemas import AgentRequest
from agentscope_runtime.engine.deployers import LocalDeployManager

print("✅ Dependencies imported successfully")

Step 2: Create Lifecycle Functions¶

Lifecycle functions define the actions to be performed when the service starts (such as state management and session history services) and how resources are released during shutdown.

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage resources during service startup and shutdown"""
    # Startup: Initialize Session manager
    import fakeredis

    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    # NOTE: This FakeRedis instance is for development/testing only.
    # In production, replace it with your own Redis client/connection
    # (e.g., aioredis.Redis)
    app.state.session = RedisSession(connection_pool=fake_redis.connection_pool)

    yield  # Service is running

    # Shutdown: Clean up logic (e.g., closing database connections) can be added here
    print("AgentApp is shutting down...")

Step 3: Create the Agent App¶

AgentApp is the core of the entire Agent application’s lifecycle and request handling, managing the application lifecycle and the registration of all services.

agent_app = AgentApp(
    app_name="Friday",
    app_description="A helpful assistant",
    lifespan=lifespan, # Pass the lifespan function
)

print("✅ Agent App created successfully")

Step 4: Define the AgentScope Query Logic¶

Important

⚠️ Important The Agent setup shown here (model, tools, conversation memory, formatter, etc.) is provided as an example configuration only. Please adapt and replace these components with your own implementations based on your requirements.

When the agent endpoint is invoked, we:

Load session context to keep different sessions isolated.
Build an Agent: includes the model, tools (such as executing Python code), conversation memory modules, and formatter.
Stream responses via stream_printing_messages, yielding (msg, last) so clients receive output as it is generated.
Persist state so the next request can resume.

@agent_app.query(framework="agentscope")
async def query_func(
    self,
    msgs,
    request: AgentRequest = None,
    **kwargs,
):
    session_id = request.session_id
    user_id = request.user_id

    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)

    agent = ReActAgent(
        name="Friday",
        model=DashScopeChatModel(
            "qwen-turbo",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            stream=True,
        ),
        sys_prompt="You're a helpful assistant named Friday.",
        toolkit=toolkit,
        memory=InMemoryMemory(),
        formatter=DashScopeChatFormatter(),
    )
    agent.set_console_output_enabled(enabled=False)

    await agent_app.state.session.load_session_state(
        session_id=session_id,
        user_id=user_id,
        agent=agent,
    )

    async for msg, last in stream_printing_messages(
        agents=[agent],
        coroutine_task=agent(msgs),
    ):
        yield msg, last

    await agent_app.state.session.save_session_state(
        session_id=session_id,
        user_id=user_id,
        agent=agent,
    )

Step 5: Run the Agent App¶

Start the Agent API server. After launch, it listens on http://localhost:8090/process:

# Start the service (listen on port 8090)
agent_app.run(host="0.0.0.0", port=8090)

# If you want to enable the built-in web chat interface at the same time, set web_ui=True
# agent_app.run(host="0.0.0.0", port=8090, web_ui=True)

Step 6: Send a Request¶

Use curl to post JSON input and observe SSE output:

curl -N \
  -X POST "http://localhost:8090/process" \
  -H "Content-Type: application/json" \
  -d '{
    "input": [
      {
        "role": "user",
        "content": [
          { "type": "text", "text": "What is the capital of France?" }
        ]
      }
    ]
  }'

The response is streamed as Server-Sent Events (SSE):

data: {"sequence_number":0,"object":"response","status":"created", ... }
data: {"sequence_number":1,"object":"response","status":"in_progress", ... }
data: {"sequence_number":2,"object":"message","status":"in_progress", ... }
data: {"sequence_number":3,"object":"content","status":"in_progress","text":"The" }
data: {"sequence_number":4,"object":"content","status":"in_progress","text":" capital of France is Paris." }
data: {"sequence_number":5,"object":"message","status":"completed","text":"The capital of France is Paris." }
data: {"sequence_number":6,"object":"response","status":"completed", ... }

Step 7: Deploy with `DeployManager`¶

AgentScope Runtime ships with a powerful deployment system. The snippet below uses LocalDeployManager:

async def main():
    await agent_app.deploy(LocalDeployManager(host="0.0.0.0", port=8091))

This spins up the agent server on the specified port so external clients can call it. Beyond the basic HTTP API, you can also interact with the deployment via protocols such as A2A, Response API, and Agent API—see Agent API Protocol Specification for details.

Clients can even call the service through the OpenAI SDK:

from openai import OpenAI

client = OpenAI(base_url="http://0.0.0.0:8091/compatible-mode/v1")

response = client.responses.create(
  model="any_name",
  input="杭州天气如何？"
)

print(response)