Xyzen 架构设计

/xyzen/api/v1/
  ├── /agents          # 代理管理
  ├── /sessions        # 会话管理
  ├── /topics          # 主题管理
  ├── /mcps            # MCP 服务器管理
  ├── /providers       # 提供者管理
  └── /auth            # 身份认证

/xyzen/ws/v1/
  └── /chat/[session_id]  # 实时聊天连接

/xyzen/api/health  # 服务健康状态检查

# FastAPI 应用初始化流程
app = FastAPI(
    title="Xyzen FastAPI Service",
    lifespan=lifespan,  # 生命周期管理
    docs_url="/xyzen/api/docs",
)

# CORS 中间件
app.add_middleware(CORSMiddleware, allow_origins=["*"])

# 根路由器
app.include_router(root_router, prefix="/xyzen")

# MCP 路由自动注册
app.router.routes.extend(setup_mcp_routes(app.state))

class Agent(SQLModel, table=True):
    id: UUID = Field(default_factory=uuid4, primary_key=True)
    name: str
    description: str | None = None
    temperature: float | None = None
    prompt: str | None = None
    user_id: str = Field(index=True)
    provider_id: UUID | None = Field(index=True)
    require_tool_confirmation: bool = False
    created_at: datetime  # TIMESTAMP with timezone
    updated_at: datetime  # TIMESTAMP with timezone

class Agent(SQLModel, table=True):
    # 基本信息
    name: str              # 代理名称
    description: str       # 代理描述
    avatar: str            # 头像 URL
    tags: list[str]        # 标签分类
    
    # 配置
    model: str             # 使用的 LLM 模型
    temperature: float     # 温度参数 (0-1)
    prompt: str            # 系统提示词
    
    # 关联
    user_id: str           # 所有者用户 ID
    provider_id: UUID      # 关联的提供者 UUID
    
    # 工具确认
    require_tool_confirmation: bool  # 调用工具前是否需要用户确认
    
    # 时间戳
    created_at: datetime   # 创建时间
    updated_at: datetime   # 更新时间

# 创建流程
POST /sessions
  └─ 自动创建一个 Topic
      └─ Topic 包含 Messages

class GraphAgent(SQLModel, table=True):
    name: str
    description: str
    state_schema: dict    # JSON - 定义状态结构
    is_active: bool       # 是否激活
    is_published: bool    # 是否发布
    is_official: bool     # 是否为官方代理
    parent_agent_id: UUID # 版本控制 - 指向上一版本

class ConsumeRecord(SQLModel, table=True):
    user_id: str           # 用户 ID
    amount: int            # 消费数量
    auth_provider: str     # 认证提供者
    
    # 业务关联
    session_id: UUID       # 关联的会话
    topic_id: UUID         # 关联的主题
    message_id: UUID       # 关联的消息
    
    # 计费状态
    consume_state: str     # pending/success/failed
    remote_response: str   # 远程计费系统响应
    
    # 幂等性
    biz_no: int           # 业务编号（自增，用于去重）

class AgentCreate(SQLModel):
    name: str
    description: str | None = None
    model: str | None = None
    temperature: float | None = None
    provider_id: UUID | None = None
    mcp_server_ids: list[UUID] = []

class AgentRead(AgentBase):
    id: UUID
    created_at: datetime
    updated_at: datetime
    # 关联数据
    mcp_servers: list[MCPServerRead]
    provider: ProviderRead

# 基础镜像: Python 3.13.5-slim
FROM python:3.13.5-slim

# 依赖管理: uv (快速 Python 包管理器)
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

# 环境变量
ENV TZ=Asia/Shanghai
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# 安装依赖
RUN uv sync --locked

# 暴露端口
EXPOSE 48196

# 启动命令
CMD ["uv", "run", "python", "-m", "app.main"]

# 数据库
DATABASE_URL=postgresql://user:password@host:5432/xyzen

# LLM 提供者
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...

# 认证
CASDOOR_ENDPOINT=https://...
BOHRIUM_API_KEY=...

# 应用
DEBUG=false
HOST=0.0.0.0
PORT=48196

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    return validate_token(token)

@router.get("/agents")
async def list_agents(current_user: User = Depends(get_current_user)):
    return await agent_service.list_agents(current_user.id)