AgentScope Runtime: A Production-grade Runtime for Agent Applications

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

# 1. Define lifespan manager
@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: Add cleanup logic here (e.g., closing database connections)
    print("AgentApp is shutting down...")

# 2. Create AgentApp instance
agent_app = AgentApp(
    app_name="Friday",
    app_description="A helpful assistant",
    lifespan=lifespan,
)

# 3. Define request handling logic
@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)

    # Load state
    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

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

# 4. Run the application
agent_app.run(host="127.0.0.1", port=8090)

# --- Synchronous version ---
from agentscope_runtime.sandbox import BaseSandbox

with BaseSandbox() as box:
    # By default, pulls `agentscope/runtime-sandbox-base:latest` from DockerHub
    print(box.list_tools()) # List all available tools
    print(box.run_ipython_cell(code="print('hi')"))  # Run Python code
    print(box.run_shell_command(command="echo hello"))  # Run shell command
    input("Press Enter to continue...")

# --- Asynchronous version ---
from agentscope_runtime.sandbox import BaseSandboxAsync

async with BaseSandboxAsync() as box:
    # Default image is `agentscope/runtime-sandbox-base:latest`
    print(await box.list_tools_async())  # List all available tools
    print(await box.run_ipython_cell(code="print('hi')"))  # Run Python code
    print(await box.run_shell_command(command="echo hello"))  # Run shell command
    input("Press Enter to continue...")

# --- Synchronous version ---
from agentscope_runtime.sandbox import GuiSandbox

with GuiSandbox() as box:
    # By default, pulls `agentscope/runtime-sandbox-gui:latest` from DockerHub
    print(box.list_tools())  # List all available tools
    print(box.desktop_url)  # Web desktop access URL
    print(box.computer_use(action="get_cursor_position"))  # Get mouse cursor position
    print(box.computer_use(action="get_screenshot"))  # Capture screenshot
    input("Press Enter to continue...")

# --- Asynchronous version ---
from agentscope_runtime.sandbox import GuiSandboxAsync

async with GuiSandboxAsync() as box:
    # Default image is `agentscope/runtime-sandbox-gui:latest`
    print(await box.list_tools_async())  # List all available tools
    print(box.desktop_url)  # Web desktop access URL
    print(await box.computer_use(action="get_cursor_position"))  # Get mouse cursor position
    print(await box.computer_use(action="get_screenshot"))  # Capture screenshot
    input("Press Enter to continue...")

# --- Synchronous version ---
from agentscope_runtime.sandbox import BrowserSandbox

with BrowserSandbox() as box:
    # By default, pulls `agentscope/runtime-sandbox-browser:latest` from DockerHub
    print(box.list_tools())  # List all available tools
    print(box.desktop_url)  # Web desktop access URL
    box.browser_navigate("https://www.google.com/")  # Open a webpage
    input("Press Enter to continue...")

# --- Asynchronous version ---
from agentscope_runtime.sandbox import BrowserSandboxAsync

async with BrowserSandboxAsync() as box:
    # Default image is `agentscope/runtime-sandbox-browser:latest`
    print(await box.list_tools_async())  # List all available tools
    print(box.desktop_url)  # Web desktop access URL
    await box.browser_navigate("https://www.google.com/")  # Open a webpage
    input("Press Enter to continue...")

# --- Synchronous version ---
from agentscope_runtime.sandbox import FilesystemSandbox

with FilesystemSandbox() as box:
    # By default, pulls `agentscope/runtime-sandbox-filesystem:latest` from DockerHub
    print(box.list_tools())  # List all available tools
    print(box.desktop_url)  # Web desktop access URL
    box.create_directory("test")  # Create a directory
    input("Press Enter to continue...")

# --- Asynchronous version ---
from agentscope_runtime.sandbox import FilesystemSandboxAsync

async with FilesystemSandboxAsync() as box:
    # Default image is `agentscope/runtime-sandbox-filesystem:latest`
    print(await box.list_tools_async())  # List all available tools
    print(box.desktop_url)  # Web desktop access URL
    await box.create_directory("test")  # Create a directory
    input("Press Enter to continue...")

# --- Synchronous version ---
from agentscope_runtime.sandbox import MobileSandbox

with MobileSandbox() as box:
    # By default, pulls 'agentscope/runtime-sandbox-mobile:latest' from DockerHub
    print(box.list_tools())  # List all available tools
    print(box.mobile_get_screen_resolution())  # Get the screen resolution
    print(box.mobile_tap([500, 1000]))  # Tap at coordinate (500, 1000)
    print(box.mobile_input_text("Hello from AgentScope!"))  # Input text
    print(box.mobile_key_event(3))  # HOME key event
    screenshot_result = box.mobile_get_screenshot()  # Get screenshot
    print(screenshot_result)
    input("Press Enter to continue...")

# --- Asynchronous version ---
from agentscope_runtime.sandbox import MobileSandboxAsync

async with MobileSandboxAsync() as box:
    # Default image is 'agentscope/runtime-sandbox-mobile:latest'
    print(await box.list_tools_async())  # List all available tools
    print(await box.mobile_get_screen_resolution())  # Get the screen resolution
    print(await box.mobile_tap([500, 1000]))  # Tap at coordinate (500, 1000)
    print(await box.mobile_input_text("Hello from AgentScope!"))  # Input text
    print(await box.mobile_key_event(3))  # HOME key event
    screenshot_result = await box.mobile_get_screenshot()  # Get screenshot
    print(screenshot_result)
    input("Press Enter to continue...")

from agentscope_runtime.engine.deployers import LocalDeployManager

# Create deployment manager
deployer = LocalDeployManager(
    host="0.0.0.0",
    port=8090,
)

# Deploy the app as a streaming service
deploy_result = await app.deploy(
    deployer=deployer,
    endpoint_path="/process"
)

from openai import OpenAI

client = OpenAI(base_url="http://localhost:8090/compatible-mode/v1")

response = client.