AgentScope Runtime

1.1.3 · active · verified Thu Apr 16

AgentScope Runtime is a production-ready framework for deploying agent applications, providing secure sandboxed execution environments, scalable deployment solutions (local, Kubernetes, serverless), and multi-framework support. It exposes agents as streaming, production-ready APIs with full-stack observability. The current version is 1.1.3, and it maintains an active development and release cadence.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to create a basic agent application using `AgentScope Runtime` to expose an AgentScope `ReActAgent` as a streaming API. It includes lifecycle management, session handling (using `fakeredis` for local testing), and uses a DashScope model (requires `DASHSCOPE_API_KEY`). To run, save as `main.py` and execute `uvicorn main:agent_app --reload --port 8090`.

import os
import uvicorn
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.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.pipeline import stream_printing_messages
import fakeredis # For local development/testing

# Configure API key (replace with your actual key or other LLM provider config)
# os.environ['DASHSCOPE_API_KEY'] = 'YOUR_API_KEY_HERE'

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize Session manager (using fakeredis for demonstration)
    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    app.state.session = RedisSession(connection_pool=fake_redis.connection_pool)
    print("\n🚀 AgentApp starting up...")
    yield
    print("\n🛑 AgentApp shutting down...")

agent_app = AgentApp(lifespan=lifespan)

@agent_app.query(framework="agentscope")
async def query_agent(msgs, request: AgentRequest = None, **kwargs):
    # Example agent setup - customize with your model, tools, and memory
    dashscope_api_key = os.environ.get('DASHSCOPE_API_KEY', 'sk-example') # Use actual key in production
    if not dashscope_api_key or dashscope_api_key == 'sk-example':
        print("Warning: DASHSCOPE_API_KEY not set. Using dummy key. Model interactions may fail.")

    model = DashScopeChatModel(
        model_name="qwen-plus", # Or your preferred model
        api_key=dashscope_api_key,
        # Other model parameters like `temperature`, `max_tokens` can be set here
    )
    agent = ReActAgent(
        name="SimpleAgent",
        model=model,
        memory=InMemoryMemory(), # Or RedisSession for persistence
        formatter=DashScopeChatFormatter(),
    )

    # Load session context (if using RedisSession in app.state.session)
    if hasattr(request, 'session_id') and app.state.session:
        session_id = request.session_id
        # In a real scenario, you'd load context using session_id
        # For this simple example, we'll just acknowledge it.
        print(f"Processing request for session_id: {session_id}")

    # Stream responses
    async for msg, last in stream_printing_messages(agent.reply(msgs, **kwargs)):
        yield msg, last


# To run this: save as e.g., `main.py` and run `uvicorn main:agent_app --reload --port 8090`
# Test with curl:
# 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?" } ] } ], "session_id": "session_1" }'

view raw JSON →