架构

curl -X POST http://localhost:8000/chat \
  -H "X-User-ID: alice" \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "agent-xxx",
    "session_id": "session-xxx",
    "input": {
      "name": "alice",
      "role": "user",
      "content": [{"type": "text", "text": "Hello"}]
    }
  }'

curl -N -H "X-User-ID: alice" \
  "http://localhost:8000/sessions/session-xxx/stream?agent_id=agent-xxx"

POST /schedule

curl -X POST -H "X-User-ID: alice" \
  "http://localhost:8000/sessions/session-xxx/interrupt?agent_id=agent-xxx"

from fastapi.middleware import Middleware
from agentscope.app import create_app
from agentscope.app.middleware import AGUIProtocolMiddleware

app = create_app(
    storage=storage,
    extra_middlewares=[
        Middleware(AGUIProtocolMiddleware),
    ],
)

from agentscope.app.middleware import ProtocolMiddlewareBase
from agentscope.event import AgentEvent

class MyProtocolMiddleware(ProtocolMiddlewareBase):
    def _convert_to_protocol(self, event: AgentEvent) -> dict:
        # 把 AgentEvent 转换为目标协议的帧格式。
        return {"type": event.type, "data": event.model_dump()}

from fastapi import Header, HTTPException, status

async def get_current_user_id(
    authorization: str = Header(...),
) -> str:
    try:
        payload = decode_jwt(authorization.removeprefix("Bearer "))
        return payload["sub"]
    except InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication token.",
        )

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user_id(token: str = Depends(oauth2_scheme)) -> str:
    user = await verify_oauth_token(token)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return user.id

from agentscope.app.deps import get_current_user_id as default_dependency

app.dependency_overrides[default_dependency] = get_current_user_id

from agentscope.app.workspace_manager import WorkspaceManagerBase
from agentscope.workspace import WorkspaceBase


class CustomWorkspaceManager(WorkspaceManagerBase):
    async def get_workspace(
        self,
        user_id: str,
        agent_id: str,
        session_id: str,
        workspace_id: str,
    ) -> WorkspaceBase:
        # 用自己的隔离策略取出已初始化的工作区。
        ...

    async def close(self, workspace_id: str) -> None:
        # 关闭并淘汰单个工作区。
        ...

    async def close_all(self) -> None:
        # 关闭全部缓存中的工作区；应用关闭时调用。
        ...

from agentscope.credential import CredentialBase
from agentscope.model import ChatModelBase

class MyProviderChatModel(ChatModelBase):
    # 针对 provider API 实现流式聊天接口。
    ...

class MyProviderCredential(CredentialBase):
    api_key: str
    endpoint: str = "https://api.my-provider.com"

    @classmethod
    def get_chat_model_class(cls):
        return MyProviderChatModel

app = create_app(
    storage=storage,
    extra_credentials=[MyProviderCredential],
)

pip install "agentscope[sql]"
pip install aiosqlite          # SQLite
# pip install asyncpg          # PostgreSQL
# pip install asyncmy          # MySQL

from agentscope.app.storage import StorageBase


class PostgresStorage(StorageBase):
    async def __aenter__(self):
        # 打开连接池。
        ...

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 关闭连接池。
        ...

    # 为每类记录实现 CRUD 方法：
    # agent、session、credential、message、schedule、team、
    # knowledge base、knowledge document。
    ...

app = create_app(
    storage=PostgresStorage(dsn="postgresql://..."),
    message_bus=message_bus,
    workspace_manager=workspace_manager,
)