AutoGPT-01-后端API模块

graph TB
    subgraph "AutoGPT后端API模块架构"
        subgraph "应用入口层 - Application Entry"
            MainApp[FastAPI主应用]
            ExternalApp[外部API应用]
            HealthCheck[健康检查端点]
        end
        
        subgraph "中间件层 - Middleware Layer"
            SecurityMW[安全头中间件]
            GZipMW[压缩中间件]
            CORSMW[跨域中间件]
            AuthMW[认证中间件]
            RateLimitMW[限流中间件]
        end
        
        subgraph "路由层 - Router Layer"
            V1Router[V1路由 /api/v1]
            V2StoreRouter[V2商店路由 /api/store]
            V2BuilderRouter[V2构建器路由 /api/builder]
            V2LibraryRouter[V2库路由 /api/library]
            V2AdminRouter[V2管理路由 /api/admin]
            IntegrationRouter[集成路由 /api/integrations]
            ExternalRouter[外部API路由 /external-api]
        end
        
        subgraph "业务处理层 - Business Logic Layer"
            AuthHandler[认证处理器]
            GraphHandler[图管理处理器]
            ExecutionHandler[执行处理器]
            StoreHandler[商店处理器]
            BlockHandler[Block处理器]
            UserHandler[用户处理器]
            IntegrationHandler[集成处理器]
        end
        
        subgraph "数据访问层 - Data Access Layer"
            GraphDB[图数据库访问]
            ExecutionDB[执行数据库访问]
            UserDB[用户数据库访问]
            StoreDB[商店数据库访问]
            CacheLayer[缓存层]
        end
        
        subgraph "异常处理层 - Exception Handling"
            ValidationHandler[验证异常处理器]
            AuthExceptionHandler[认证异常处理器]
            BusinessExceptionHandler[业务异常处理器]
            SystemExceptionHandler[系统异常处理器]
        end
    end
    
    %% 连接关系
    MainApp --> SecurityMW
    SecurityMW --> GZipMW
    GZipMW --> CORSMW
    CORSMW --> AuthMW
    AuthMW --> RateLimitMW
    
    RateLimitMW --> V1Router
    RateLimitMW --> V2StoreRouter
    RateLimitMW --> V2BuilderRouter
    RateLimitMW --> V2LibraryRouter
    RateLimitMW --> V2AdminRouter
    RateLimitMW --> IntegrationRouter
    
    V1Router --> AuthHandler
    V1Router --> GraphHandler
    V1Router --> ExecutionHandler
    V1Router --> BlockHandler
    
    V2StoreRouter --> StoreHandler
    V2LibraryRouter --> GraphHandler
    V2AdminRouter --> UserHandler
    IntegrationRouter --> IntegrationHandler
    
    AuthHandler --> UserDB
    GraphHandler --> GraphDB
    ExecutionHandler --> ExecutionDB
    StoreHandler --> StoreDB
    BlockHandler --> CacheLayer
    
    MainApp --> ValidationHandler
    MainApp --> AuthExceptionHandler
    MainApp --> BusinessExceptionHandler
    MainApp --> SystemExceptionHandler

# /autogpt_platform/backend/backend/server/rest_api.py

import fastapi
import starlette.middleware.cors
import uvicorn
from autogpt_libs.auth import add_auth_responses_to_openapi
from fastapi.middleware.gzip import GZipMiddleware
from backend.server.middleware.security import SecurityHeadersMiddleware

# 创建FastAPI应用实例
app = fastapi.FastAPI(
    title="AutoGPT Server",
    description="AutoGPT Server API",
    summary="AutoGPT Server API",
    version="0.1",
    openapi_url="/openapi.json",
    docs_url="/docs",
    redoc_url="/redoc",
)

# 中间件配置 - 按执行顺序添加
app.add_middleware(SecurityHeadersMiddleware)  # 安全头中间件
app.add_middleware(GZipMiddleware, minimum_size=1000)  # 压缩中间件

# CORS配置 - 支持跨域请求
app.add_middleware(
    starlette.middleware.cors.CORSMiddleware,
    allow_origins=["*"],  # 生产环境应配置具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# OpenAPI认证响应配置
add_auth_responses_to_openapi(app)

# 路由挂载 - 按功能模块和版本组织
app.include_router(backend.server.routers.v1.v1_router, tags=["v1"], prefix="/api")
app.include_router(
    backend.server.v2.store.routes.router, tags=["v2"], prefix="/api/store"
)
app.include_router(
    backend.server.v2.builder.routes.router, tags=["v2"], prefix="/api/builder"
)
app.include_router(
    backend.server.v2.library.routes.router, tags=["v2"], prefix="/api/library"
)
app.include_router(
    backend.server.v2.admin.store_admin_routes.router,
    tags=["v2", "admin"],
    prefix="/api/store",
)

# 外部API挂载为独立应用
app.mount("/external-api", external_app)

# 异常处理器注册 - 统一错误响应格式
app.add_exception_handler(PrismaError, handle_internal_http_error(500))
app.add_exception_handler(NotFoundError, handle_internal_http_error(404, False))
app.add_exception_handler(NotAuthorizedError, handle_internal_http_error(403, False))
app.add_exception_handler(RequestValidationError, validation_error_handler)
app.add_exception_handler(pydantic.ValidationError, validation_error_handler)
app.add_exception_handler(ValueError, handle_internal_http_error(400))
app.add_exception_handler(Exception, handle_internal_http_error(500))

def handle_internal_http_error(status_code: int, log_error: bool = True):
    """创建HTTP错误处理器的工厂函数"""
    async def handler(request: fastapi.Request, exc: Exception):
        if log_error:
            logger.exception(f"Internal error: {exc}")
        
        return fastapi.responses.JSONResponse(
            status_code=status_code,
            content={
                "detail": str(exc),
                "error_type": type(exc).__name__,
                "timestamp": datetime.utcnow().isoformat(),
            }
        )
    return handler

# /autogpt_platform/backend/backend/server/routers/v1.py

from fastapi import APIRouter, Security, HTTPException, Body
from autogpt_libs.auth import requires_user, requires_admin_user, get_user_id, get_jwt_payload

# 创建V1路由器
v1_router = APIRouter()

# 集成子路由器
v1_router.include_router(
    backend.server.integrations.router.router,
    prefix="/integrations",
    tags=["integrations"],
)

v1_router.include_router(
    backend.server.routers.analytics.router,
    prefix="/analytics", 
    tags=["analytics"],
    dependencies=[Security(requires_user)],
)

@v1_router.post(
    "/auth/user",
    summary="Get or create user",
    tags=["auth"],
    dependencies=[Security(requires_user)],
)
async def get_or_create_user_route(user_data: dict = Security(get_jwt_payload)):
    """
    获取或创建用户端点
    
    功能说明:
    1. 从JWT令牌中提取用户信息
    2. 检查用户是否存在于数据库中
    3. 如果用户不存在则创建新用户记录
    4. 返回用户的完整信息
    
    认证要求:
    - 需要有效的JWT令牌
    - 令牌必须包含用户基本信息（sub、email等）
    
    参数:
        user_data: 从JWT令牌中提取的用户数据字典
        
    返回:
        用户信息字典，包含id、email、name等字段
    """
    user = await get_or_create_user(user_data)
    return user.model_dump()

{
    "id": "uuid",           // 用户唯一标识符
    "email": "string",      // 用户邮箱地址
    "name": "string",       // 用户显示名称
    "role": "string",       // 用户角色（user/admin）
    "timezone": "string",   // 用户时区
    "created_at": "string", // 创建时间（ISO格式）
    "updated_at": "string"  // 更新时间（ISO格式）
}

sequenceDiagram
    participant Client as 客户端
    participant Router as V1Router
    participant Auth as 认证中间件
    participant Handler as get_or_create_user_route
    participant UserService as get_or_create_user
    participant DB as 数据库

    Client->>Router: POST /api/v1/auth/user
    Router->>Auth: Security(requires_user)
    Auth->>Auth: get_jwt_payload(credentials)
    Auth->>Auth: parse_jwt_token(token)
    Auth->>Auth: verify_user(payload)
    Auth-->>Router: 用户数据字典
    Router->>Handler: 调用处理函数
    Handler->>UserService: get_or_create_user(user_data)
    UserService->>DB: 查询用户是否存在
    alt 用户存在
        DB-->>UserService: 返回现有用户
        UserService->>DB: 更新用户信息
    else 用户不存在
        UserService->>DB: 创建新用户记录
    end
    DB-->>UserService: 用户对象
    UserService-->>Handler: User模型
    Handler-->>Client: 用户信息字典

# /autogpt_platform/backend/backend/data/user.py

async def get_or_create_user(user_data: dict) -> User:
    """
    获取或创建用户的核心业务逻辑
    
    处理流程:
    1. 从JWT载荷中提取用户标识和基本信息
    2. 查询数据库检查用户是否已存在
    3. 如果用户存在，更新其基本信息（邮箱、名称等）
    4. 如果用户不存在，创建新的用户记录
    5. 返回用户对象供API响应使用
    
    参数:
        user_data: JWT载荷字典，包含以下字段：
            - sub: 用户唯一标识符（Supabase用户ID）
            - email: 用户邮箱地址
            - user_metadata: 用户元数据，可能包含name等信息
            - created_at: 账户创建时间
            - last_sign_in_at: 最后登录时间
    
    返回:
        User: 用户模型对象，包含完整的用户信息
    
    异常:
        DatabaseError: 数据库操作失败时抛出
        ValidationError: 用户数据验证失败时抛出
    """
    user_id = user_data["sub"]
    email = user_data["email"]
    user_metadata = user_data.get("user_metadata", {})
    
    # 查询现有用户
    existing_user = await prisma.user.find_unique(where={"id": user_id})
    
    if existing_user:
        # 更新现有用户信息
        updated_user = await prisma.user.update(
            where={"id": user_id},
            data={
                "email": email,
                "name": user_metadata.get("name"),
                "updated_at": datetime.utcnow(),
            }
        )
        return updated_user
    else:
        # 创建新用户
        new_user = await prisma.user.create(
            data={
                "id": user_id,
                "email": email,
                "name": user_metadata.get("name"),
                "created_at": datetime.utcnow(),
                "updated_at": datetime.utcnow(),
            }
        )
        return new_user

@v1_router.post(
    "/auth/user/email",
    summary="Update user email",
    tags=["auth"],
    dependencies=[Security(requires_user)],
)
async def update_user_email_route(
    user_id: Annotated[str, Security(get_user_id)], 
    email: str = Body(...)
) -> dict[str, str]:
    """
    更新用户邮箱端点
    
    功能说明:
    1. 验证用户身份和权限
    2. 验证新邮箱地址的格式和唯一性
    3. 更新数据库中的用户邮箱
    4. 返回更新确认信息
    
    参数:
        user_id: 从JWT令牌中提取的用户ID
        email: 新的邮箱地址（从请求体中获取）
        
    返回:
        包含新邮箱地址的确认字典
    """
    await update_user_email(user_id, email)
    return {"email": email}

{
    "email": "new.email@example.com"  // 新的邮箱地址，必须符合邮箱格式
}

{
    "email": "new.email@example.com"  // 更新后的邮箱地址
}

@v1_router.post(
    path="/graphs",
    summary="Create new graph",
    tags=["graphs"],
    dependencies=[Security(requires_user)],
)
async def create_new_graph(
    create_graph: CreateGraph,
    user_id: Annotated[str, Security(get_user_id)],
) -> graph_db.GraphModel:
    """
    创建新图端点
    
    功能说明:
    1. 验证用户权限和输入数据
    2. 为图和节点分配新的UUID
    3. 验证图结构的完整性和正确性
    4. 创建图记录和相关的节点记录
    5. 在库中创建对应的智能体记录
    6. 激活新创建的图版本
    
    参数:
        create_graph: 图创建请求对象，包含图定义
        user_id: 创建者的用户ID
        
    返回:
        GraphModel: 创建成功的图模型对象
    """
    # 创建图模型并分配ID
    graph = graph_db.make_graph_model(create_graph.graph, user_id)
    graph.reassign_ids(user_id=user_id, reassign_graph_id=True)
    
    # 验证图结构
    graph.validate_graph(for_run=False)

    # 创建图和库记录
    await graph_db.create_graph(graph, user_id=user_id)
    await library_db.create_library_agent(graph, user_id=user_id)
    
    # 激活图
    return await on_graph_activate(graph, user_id=user_id)

{
    "graph": {
        "name": "string",           // 图名称，最大100字符
        "description": "string",    // 图描述，可选
        "nodes": [                  // 节点列表
            {
                "id": "string",     // 节点ID，可为空（自动生成）
                "block_id": "string", // Block类型ID
                "input_default": {},  // 默认输入值
                "metadata": {        // 节点元数据
                    "position": {"x": 0, "y": 0}  // 节点位置
                }
            }
        ],
        "links": [                  // 连接列表
            {
                "source_id": "string",   // 源节点ID
                "sink_id": "string",     // 目标节点ID
                "source_name": "string", // 源输出端口名
                "sink_name": "string"    // 目标输入端口名
            }
        ]
    }
}

@v1_router.post(
    path="/graphs/{graph_id}/execute/{graph_version}",
    summary="Execute graph agent",
    tags=["graphs"],
    dependencies=[Security(requires_user)],
)
async def execute_graph(
    graph_id: str,
    user_id: Annotated[str, Security(get_user_id)],
    inputs: Annotated[dict[str, Any], Body(..., embed=True, default_factory=dict)],
    credentials_inputs: Annotated[
        dict[str, CredentialsMetaInput], Body(..., embed=True, default_factory=dict)
    ],
    graph_version: Optional[int] = None,
    preset_id: Optional[str] = None,
) -> execution_db.GraphExecutionMeta:
    """
    执行图的核心端点
    
    功能说明:
    1. 验证用户积分余额是否充足
    2. 获取指定版本的图定义
    3. 验证图结构和输入数据
    4. 创建执行记录并分配执行ID
    5. 将执行任务提交到任务队列
    6. 返回执行元数据供状态跟踪
    
    参数:
        graph_id: 要执行的图的唯一标识符
        user_id: 执行用户的ID，用于权限验证和计费
        inputs: 图执行的输入数据字典
        credentials_inputs: 执行所需的第三方服务凭据
        graph_version: 指定执行的图版本，默认使用最新版本
        preset_id: 预设配置ID，用于加载预定义参数
        
    返回:
        GraphExecutionMeta: 执行元数据，包含执行ID和状态
    """
    # 检查用户积分余额
    current_balance = await _user_credit_model.get_credits(user_id)
    if current_balance <= 0:
        raise HTTPException(
            status_code=402,
            detail="Insufficient balance to execute the agent. Please top up your account.",
        )

    try:
        # 调用执行工具函数
        result = await execution_utils.add_graph_execution(
            graph_id=graph_id,
            user_id=user_id,
            inputs=inputs,
            preset_id=preset_id,
            graph_version=graph_version,
            graph_credentials_inputs=credentials_inputs,
        )
        
        return result
        
    except GraphValidationError as e:
        raise HTTPException(
            status_code=400,
            detail={
                "type": "validation_error",
                "message": e.message,
                "node_errors": e.node_errors,
            },
        )

{
    "inputs": {                     // 图执行输入数据
        "input_field": "value"      // 键值对形式的输入
    },
    "credentials_inputs": {         // 凭据输入配置
        "provider_name": {
            "type": "api_key",      // 凭据类型
            "id": "credential_id"   // 凭据ID
        }
    }
}

{
    "id": "uuid",                   // 执行唯一标识符
    "graph_id": "uuid",             // 图ID
    "graph_version": 1,             // 图版本号
    "user_id": "uuid",              // 执行用户ID
    "status": "QUEUED",             // 执行状态
    "created_at": "2025-01-27T10:00:00Z",  // 创建时间
    "started_at": null,             // 开始时间（初始为null）
    "ended_at": null,               // 结束时间（初始为null）
    "inputs": {},                   // 输入数据
    "outputs": {}                   // 输出数据（执行完成后填充）
}

sequenceDiagram
    participant Client as 客户端
    participant API as 执行API
    participant Auth as 认证服务
    participant Credit as 积分服务
    participant GraphDB as 图数据库
    participant ExecUtils as 执行工具
    participant Queue as 执行队列
    participant Manager as 执行管理器

    Client->>API: POST /graphs/{id}/execute
    API->>Auth: Security(requires_user)
    Auth->>Auth: 验证JWT令牌
    Auth-->>API: 用户ID

    API->>Credit: 检查用户积分
    Credit->>Credit: 查询用户余额
    Credit-->>API: 余额信息

    alt 余额不足
        API-->>Client: HTTP 402 余额不足
    else 余额充足
        API->>ExecUtils: add_graph_execution()
        ExecUtils->>GraphDB: 获取图定义
        GraphDB-->>ExecUtils: 图结构数据
        ExecUtils->>ExecUtils: 验证图结构
        ExecUtils->>ExecUtils: 创建执行记录
        ExecUtils->>Queue: 提交执行任务
        Queue-->>Manager: 通知新任务
        ExecUtils-->>API: 执行元数据
        API-->>Client: 执行ID和状态
    end

@v1_router.get(
    path="/blocks",
    summary="List available blocks",
    tags=["blocks"],
    dependencies=[Security(requires_user)],
)
async def get_graph_blocks() -> Sequence[dict[Any, Any]]:
    """
    获取可用Block列表端点
    
    功能说明:
    1. 从Block注册表中获取所有已加载的Block
    2. 包含每个Block的完整配置信息
    3. 计算Block的执行成本信息
    4. 过滤掉被禁用的Block
    5. 应用缓存机制提高响应性能
    
    返回:
        Block信息列表，每个Block包含schema、成本等信息
    """
    return _get_cached_blocks()

@cached()
def _get_cached_blocks() -> Sequence[dict[Any, Any]]:
    """
    获取缓存的Block列表
    
    使用缓存装饰器防止并发请求重复执行昂贵的Block加载操作
    
    处理流程:
    1. 从Block注册表获取所有Block类
    2. 为每个Block创建实例并提取信息
    3. 计算Block的执行成本
    4. 过滤禁用的Block
    5. 返回格式化的Block信息列表
    
    返回:
        格式化的Block信息字典列表
    """
    from backend.data.credit import get_block_cost

    block_classes = get_blocks()
    result = []

    for block_class in block_classes.values():
        block_instance = block_class()
        
        # 跳过禁用的Block
        if not block_instance.disabled:
            # 获取Block成本信息
            costs = get_block_cost(block_instance)
            
            # 构建Block完整信息
            block_info = {
                **block_instance.to_dict(),
                "costs": costs
            }
            result.append(block_info)

    return result

[
    {
        "id": "uuid",               // Block唯一标识符
        "name": "BlockName",        // Block类名
        "description": "string",    // Block功能描述
        "categories": [             // Block分类列表
            {
                "category": "AI",
                "description": "Block that leverages AI to perform a task."
            }
        ],
        "input_schema": {           // 输入数据schema（JSON Schema格式）
            "type": "object",
            "properties": {
                "field_name": {
                    "type": "string",
                    "description": "字段描述"
                }
            },
            "required": ["field_name"]
        },
        "output_schema": {          // 输出数据schema
            "type": "object",
            "properties": {
                "result": {
                    "type": "string",
                    "description": "处理结果"
                }
            }
        },
        "static_output": false,     // 是否为静态输出
        "block_type": "Standard",   // Block类型
        "costs": [                  // 成本配置列表
            {
                "cost_amount": 10,
                "cost_type": "run",
                "cost_filter": {}
            }
        ]
    }
]

@v1_router.post(
    path="/blocks/{block_id}/execute",
    summary="Execute graph block",
    tags=["blocks"],
    dependencies=[Security(requires_user)],
)
async def execute_graph_block(
    block_id: str, 
    data: BlockInput, 
    user_id: Annotated[str, Security(get_user_id)]
) -> CompletedBlockOutput:
    """
    执行单个Block端点
    
    功能说明:
    1. 根据block_id获取对应的Block实例
    2. 验证用户权限和输入数据
    3. 获取用户上下文信息（时区等）
    4. 执行Block的核心逻辑
    5. 收集所有输出数据
    6. 记录执行指标和统计信息
    
    参数:
        block_id: 要执行的Block的唯一标识符
        data: Block输入数据
        user_id: 执行用户的ID
        
    返回:
        CompletedBlockOutput: 完整的Block执行输出
    """
    # 获取Block对象
    obj = get_block(block_id)
    if not obj:
        raise HTTPException(status_code=404, detail=f"Block #{block_id} not found.")

    # 获取用户上下文
    user = await get_user_by_id(user_id)
    user_context = UserContext(timezone=user.timezone)

    # 执行Block
    output = defaultdict(list)
    
    async for name, data in obj.execute(
        data,
        user_context=user_context,
        user_id=user_id,
    ):
        output[name].append(data)

    return output

{
    "data": {                       // Block输入数据
        "input_field": "value",     // 根据Block的input_schema定义
        "another_field": 123
    },
    "credentials": {                // 可选的凭据信息
        "provider_name": {
            "api_key": "secret"
        }
    }
}

{
    "result": [                     // 主要输出结果列表
        "output_value_1",
        "output_value_2"
    ],
    "metadata": [                   // 元数据输出列表
        {
            "execution_time": 1.23,
            "tokens_used": 150
        }
    ],
    "error": []                     // 错误信息列表（如果有）
}

# /autogpt_platform/backend/backend/server/v2/store/routes.py

@router.get(
    "/agents",
    summary="List store agents",
    tags=["store", "public"],
    response_model=StoreAgentsResponse,
)
async def get_agents(
    featured: bool = False,
    creator: str | None = None,
    sorted_by: str | None = None,
    search_query: str | None = None,
    category: str | None = None,
    page: int = 1,
    page_size: int = 20,
):
    """
    获取商店智能体列表端点
    
    功能说明:
    1. 支持多维度筛选（特色、创建者、分类）
    2. 支持全文搜索和智能排序
    3. 实现分页机制控制响应大小
    4. 应用缓存策略提高查询性能
    5. 返回结构化的智能体信息
    
    查询参数:
        featured: 是否只显示特色智能体
        creator: 按创建者用户名筛选
        sorted_by: 排序方式（runs/rating/recent）
        search_query: 搜索关键词，支持名称和描述搜索
        category: 按分类筛选
        page: 页码，从1开始
        page_size: 每页数量，最大100
        
    返回:
        StoreAgentsResponse: 包含智能体列表和分页信息
    """
    # 参数验证
    if page < 1:
        raise fastapi.HTTPException(
            status_code=422, detail="Page must be greater than 0"
        )

    if page_size < 1:
        raise fastapi.HTTPException(
            status_code=422, detail="Page size must be greater than 0"
        )

    # 缓存查询
    agents = await _get_cached_store_agents(
        featured=featured,
        creator=creator,
        sorted_by=sorted_by,
        search_query=search_query,
        category=category,
        page=page,
        page_size=page_size,
    )
    return agents

{
    "agents": [
        {
            "id": "uuid",               // 智能体ID
            "name": "string",           // 智能体名称
            "description": "string",    // 详细描述
            "sub_heading": "string",    // 副标题
            "category": "string",       // 分类
            "rating": 4.5,              // 评分（0-5）
            "runs": 1234,               // 运行次数
            "creator": {                // 创建者信息
                "username": "string",
                "name": "string",
                "avatar_url": "string"
            },
            "image_urls": ["string"],   // 预览图片URL列表
            "video_url": "string",      // 演示视频URL
            "created_at": "string",     // 创建时间
            "updated_at": "string"      // 更新时间
        }
    ],
    "total_count": 100,                 // 总数量
    "page": 1,                          // 当前页码
    "page_size": 20,                    // 每页大小
    "total_pages": 5                    // 总页数
}

@router.post(
    "/submissions",
    summary="Create store submission",
    tags=["store", "private"],
    dependencies=[fastapi.Security(autogpt_libs.auth.requires_user)],
    response_model=StoreSubmission,
)
async def create_submission(
    submission_request: StoreSubmissionRequest,
    user_id: str = fastapi.Security(autogpt_libs.auth.get_user_id),
):
    """
    创建商店提交端点
    
    功能说明:
    1. 验证用户权限和智能体所有权
    2. 检查提交数据的完整性和唯一性
    3. 创建商店列表记录和版本记录
    4. 将提交加入审核队列
    5. 清除相关缓存确保数据一致性
    
    参数:
        submission_request: 提交请求对象
        user_id: 提交用户的ID
        
    返回:
        StoreSubmission: 创建的提交对象
    """
    result = await backend.server.v2.store.db.create_store_submission(
        user_id=user_id,
        agent_id=submission_request.agent_id,
        agent_version=submission_request.agent_version,
        slug=submission_request.slug,
        name=submission_request.name,
        video_url=submission_request.video_url,
        image_urls=submission_request.image_urls,
        description=submission_request.description,
        instructions=submission_request.instructions,
        sub_heading=submission_request.sub_heading,
        categories=submission_request.categories,
        changes_summary=submission_request.changes_summary or "Initial Submission",
        recommended_schedule_cron=submission_request.recommended_schedule_cron,
    )

    # 清除用户提交缓存
    for page in range(1, 20):
        _get_cached_submissions.cache_delete(user_id, page=page, page_size=20)

    return result

{
    "agent_id": "uuid",             // 智能体ID
    "agent_version": 1,             // 智能体版本
    "slug": "unique-slug",          // 唯一标识符（URL友好）
    "name": "智能体名称",           // 显示名称
    "description": "详细描述",      // 功能描述
    "instructions": "使用说明",     // 使用指导
    "sub_heading": "简短描述",      // 副标题
    "categories": ["AI", "TEXT"],   // 分类标签
    "video_url": "https://...",     // 演示视频URL（可选）
    "image_urls": ["https://..."],  // 预览图片URL列表
    "changes_summary": "更新说明", // 变更总结
    "recommended_schedule_cron": "0 9 * * *"  // 推荐执行计划（可选）
}

# /autogpt_platform/backend/backend/server/v2/library/routes/agents.py

@router.get(
    "/agents",
    summary="List library agents",
    tags=["library", "private"],
    dependencies=[fastapi.Security(autogpt_libs.auth.requires_user)],
    response_model=LibraryAgentsResponse,
)
async def get_library_agents(
    user_id: str = fastapi.Security(autogpt_libs.auth.get_user_id),
    page: int = 1,
    page_size: int = 20,
    search_query: str | None = None,
    sort_by: str = "created_at",
    sort_order: str = "desc",
):
    """
    获取用户库智能体列表
    
    功能说明:
    1. 获取用户私有的智能体库
    2. 支持按名称和描述搜索
    3. 支持多种排序方式
    4. 包含智能体的执行统计信息
    5. 实现分页和缓存机制
    
    参数:
        user_id: 用户ID，从JWT令牌中提取
        page: 页码
        page_size: 每页数量
        search_query: 搜索关键词
        sort_by: 排序字段（created_at/name/runs）
        sort_order: 排序方向（asc/desc）
        
    返回:
        LibraryAgentsResponse: 用户智能体列表和统计信息
    """
    return await backend.server.v2.library.db.get_library_agents(
        user_id=user_id,
        page=page,
        page_size=page_size,
        search_query=search_query,
        sort_by=sort_by,
        sort_order=sort_order,
    )

# /autogpt_platform/backend/backend/server/external/api.py

from fastapi import FastAPI
from .routes.v1 import router as v1_router

# 创建外部API应用实例
external_app = FastAPI(
    title="AutoGPT External API",
    description="External API for third-party integrations",
    version="1.0",
    docs_url="/docs",
    openapi_url="/openapi.json",
)

# 挂载V1外部路由
external_app.include_router(v1_router, prefix="/v1", tags=["external-v1"])

# /autogpt_platform/backend/backend/server/external/routes/v1.py

@router.post(
    "/graphs/{graph_id}/execute",
    summary="Execute graph via external API",
    tags=["external"],
)
async def execute_graph_external(
    graph_id: str,
    execution_request: ExternalExecutionRequest,
    api_key: str = Security(validate_api_key),
) -> ExternalExecutionResponse:
    """
    外部图执行端点
    
    功能说明:
    1. 基于API Key进行身份验证
    2. 验证API Key的权限范围
    3. 转换外部请求格式为内部格式
    4. 调用内部执行引擎
    5. 转换响应格式为外部标准
    
    参数:
        graph_id: 要执行的图ID
        execution_request: 外部执行请求
        api_key: API密钥，用于身份验证
        
    返回:
        ExternalExecutionResponse: 标准化的外部响应
    """
    # 获取API Key关联的用户
    user_id = await get_user_by_api_key(api_key)
    
    # 转换请求格式
    internal_request = convert_external_to_internal_request(execution_request)
    
    # 调用内部执行API
    execution_result = await internal_execute_graph(
        graph_id=graph_id,
        user_id=user_id,
        **internal_request
    )
    
    # 转换响应格式
    return convert_internal_to_external_response(execution_result)

# 认证调用链路核心函数

# 1. JWT载荷提取
def get_jwt_payload(
    credentials: HTTPAuthorizationCredentials | None = Security(bearer_jwt_auth),
) -> dict[str, Any]:
    """
    从HTTP Authorization头部提取JWT载荷
    
    处理流程:
    1. 检查Authorization头部是否存在
    2. 提取Bearer令牌
    3. 调用JWT解析函数
    4. 返回解析后的载荷数据
    
    参数:
        credentials: FastAPI Security依赖注入的凭据对象
        
    返回:
        JWT载荷字典，包含用户信息
        
    异常:
        HTTPException: 令牌缺失或无效时抛出401错误
    """
    if not credentials:
        raise HTTPException(
            status_code=401, 
            detail="Authorization header is missing"
        )

    try:
        payload = parse_jwt_token(credentials.credentials)
        return payload
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

# 2. JWT令牌解析
def parse_jwt_token(token: str) -> dict[str, Any]:
    """
    解析和验证JWT令牌
    
    验证步骤:
    1. 使用配置的验证密钥验证签名
    2. 检查令牌是否过期
    3. 验证audience声明
    4. 返回解码的载荷
    
    参数:
        token: 要解析的JWT令牌字符串
        
    返回:
        解码后的载荷字典
        
    异常:
        ValueError: 令牌无效、过期或签名错误时抛出
    """
    settings = get_settings()
    
    try:
        payload = jwt.decode(
            token,
            settings.JWT_VERIFY_KEY,
            algorithms=[settings.JWT_ALGORITHM],
            audience="authenticated",
        )
        return payload
        
    except jwt.ExpiredSignatureError:
        raise ValueError("Token has expired")
    except jwt.InvalidSignatureError:
        raise ValueError("Invalid token signature")
    except jwt.InvalidTokenError as e:
        raise ValueError(f"Invalid token: {str(e)}")

# 3. 用户验证
def verify_user(jwt_payload: dict | None, admin_only: bool) -> User:
    """
    验证用户身份和权限
    
    验证流程:
    1. 检查JWT载荷是否存在
    2. 提取用户ID（sub字段）
    3. 如果需要管理员权限，验证角色
    4. 创建User对象返回
    
    参数:
        jwt_payload: JWT载荷字典
        admin_only: 是否只允许管理员用户
        
    返回:
        验证成功的User对象
        
    异常:
        HTTPException: 认证失败或权限不足时抛出
    """
    if jwt_payload is None:
        raise HTTPException(
            status_code=401, 
            detail="Authorization header is missing"
        )

    user_id = jwt_payload.get("sub")
    if not user_id:
        raise HTTPException(
            status_code=401, 
            detail="User ID not found in token"
        )

    # 管理员权限检查
    if admin_only:
        user_role = jwt_payload.get("role", "").lower()
        if user_role != "admin":
            raise HTTPException(
                status_code=403, 
                detail="Admin access required"
            )

    # 创建用户对象
    user = User.from_payload(jwt_payload)
    return user

sequenceDiagram
    participant Client as 客户端
    participant FastAPI as FastAPI应用
    participant Security as Security依赖
    participant JWT as JWT工具
    participant User as 用户服务

    Client->>FastAPI: HTTP请求 + Authorization头
    FastAPI->>Security: Security(requires_user)
    Security->>Security: get_jwt_payload()
    Security->>JWT: parse_jwt_token(token)
    JWT->>JWT: 验证签名和过期时间
    JWT-->>Security: JWT载荷
    Security->>Security: verify_user(payload)
    Security->>User: User.from_payload()
    User-->>Security: User对象
    Security-->>FastAPI: 认证成功
    FastAPI->>FastAPI: 执行业务逻辑
    FastAPI-->>Client: API响应

# /autogpt_platform/backend/backend/util/execution.py

async def add_graph_execution(
    graph_id: str,
    user_id: str,
    inputs: dict[str, Any],
    preset_id: Optional[str] = None,
    graph_version: Optional[int] = None,
    graph_credentials_inputs: Optional[dict[str, CredentialsMetaInput]] = None,
) -> execution_db.GraphExecutionMeta:
    """
    添加图执行的核心业务逻辑
    
    执行步骤:
    1. 获取指定版本的图定义
    2. 验证用户对图的访问权限
    3. 处理预设配置的合并
    4. 验证图结构和输入数据
    5. 处理执行凭据配置
    6. 创建执行记录
    7. 提交到执行队列
    
    参数:
        graph_id: 图的唯一标识符
        user_id: 执行用户ID
        inputs: 图执行输入数据
        preset_id: 预设配置ID（可选）
        graph_version: 图版本号（可选，默认最新）
        graph_credentials_inputs: 凭据配置
        
    返回:
        GraphExecutionMeta: 执行元数据对象
        
    异常:
        ValueError: 图不存在或权限不足
        GraphValidationError: 图验证失败
    """
    # 获取图定义
    graph = await graph_db.get_graph(
        graph_id=graph_id,
        version=graph_version,
        user_id=user_id,
        include_subgraphs=True,
    )
    
    if not graph:
        raise ValueError(f"Graph #{graph_id} not found or access denied.")

    # 处理预设配置
    if preset_id:
        preset = await preset_db.get_preset(preset_id, user_id)
        if preset:
            inputs = {**preset.inputs, **inputs}
            if preset.credentials:
                merged_credentials = {**preset.credentials, **(graph_credentials_inputs or {})}
                graph_credentials_inputs = merged_credentials

    # 图验证
    graph.validate_graph(for_run=True)

    # 创建执行记录
    graph_exec = await execution_db.create_graph_execution(
        graph_id=graph_id,
        graph_version=graph.version,
        user_id=user_id,
        inputs=inputs,
        credentials_inputs=graph_credentials_inputs or {},
        status=execution_db.ExecutionStatus.QUEUED,
    )

    # 创建执行队列条目
    execution_entry = GraphExecutionEntry(
        id=graph_exec.id,
        graph_id=graph_id,
        graph_version=graph.version,
        user_id=user_id,
        inputs=inputs,
        credentials_inputs=graph_credentials_inputs or {},
        created_at=datetime.utcnow(),
    )
    
    # 提交到执行队列
    execution_queue = get_execution_queue()
    await execution_queue.put(execution_entry)

    return graph_exec

sequenceDiagram
    participant API as 执行API
    participant GraphDB as 图数据库
    participant PresetDB as 预设数据库
    participant Validator as 图验证器
    participant ExecDB as 执行数据库
    participant Queue as 执行队列
    participant Manager as 执行管理器

    API->>GraphDB: get_graph(graph_id, version, user_id)
    GraphDB-->>API: 图定义对象

    alt 有预设配置
        API->>PresetDB: get_preset(preset_id, user_id)
        PresetDB-->>API: 预设配置
        API->>API: 合并预设和输入数据
    end

    API->>Validator: graph.validate_graph(for_run=True)
    Validator->>Validator: 检查节点连接
    Validator->>Validator: 检查循环依赖
    Validator->>Validator: 验证输入数据
    Validator-->>API: 验证通过

    API->>ExecDB: create_graph_execution()
    ExecDB-->>API: 执行记录

    API->>Queue: put(execution_entry)
    Queue-->>Manager: 通知新执行任务
    API-->>Client: 执行元数据

# 验证异常处理器
async def validation_error_handler(
    request: fastapi.Request, exc: Exception
) -> fastapi.responses.Response:
    """
    处理输入验证异常
    
    功能说明:
    1. 捕获Pydantic和FastAPI的验证异常
    2. 提取详细的字段验证错误信息
    3. 格式化错误响应为标准结构
    4. 返回422状态码表示请求格式错误
    
    参数:
        request: FastAPI请求对象
        exc: 验证异常对象
        
    返回:
        标准化的错误响应
    """
    if isinstance(exc, RequestValidationError):
        errors = []
        for error in exc.errors():
            errors.append({
                "field": ".".join(str(loc) for loc in error["loc"]),
                "message": error["msg"],
                "type": error["type"],
                "input": error.get("input"),
            })
        
        return fastapi.responses.JSONResponse(
            content={
                "detail": "Validation error",
                "errors": errors,
                "error_type": "validation_error",
            },
            status_code=422,
        )
    
    # 处理其他验证异常
    return fastapi.responses.JSONResponse(
        content={
            "detail": str(exc),
            "error_type": type(exc).__name__,
        },
        status_code=422,
    )

{
    "detail": "错误描述信息",
    "error_type": "异常类型名称",
    "timestamp": "2025-01-27T10:00:00Z",
    "errors": [                     // 验证错误时的详细信息
        {
            "field": "字段路径",
            "message": "错误消息",
            "type": "错误类型",
            "input": "输入值"
        }
    ]
}

# 缓存装饰器应用
@cached(maxsize=5000, ttl_seconds=900)  # 15分钟TTL，最多5000个条目
async def _get_cached_store_agents(
    featured: bool,
    creator: str | None,
    sorted_by: str | None,
    search_query: str | None,
    category: str | None,
    page: int,
    page_size: int,
):
    """
    缓存的商店智能体查询
    
    缓存策略:
    1. 基于查询参数生成缓存键
    2. 15分钟TTL平衡数据新鲜度和性能
    3. LRU淘汰策略控制内存使用
    4. 查询参数变化时自动失效
    
    参数:
        各种查询筛选参数
        
    返回:
        缓存的查询结果
    """
    return await backend.server.v2.store.db.get_store_agents(
        featured=featured,
        creators=[creator] if creator else None,
        sorted_by=sorted_by,
        search_query=search_query,
        category=category,
        page=page,
        page_size=page_size,
    )

# /autogpt_platform/backend/backend/server/v2/store/db.py

async def get_store_agents(
    featured: bool = False,
    creators: Optional[List[str]] = None,
    sorted_by: Optional[str] = None,
    search_query: Optional[str] = None,
    category: Optional[str] = None,
    page: int = 1,
    page_size: int = 20,
) -> StoreAgentsResponse:
    """
    优化的商店智能体查询实现
    
    优化策略:
    1. 使用JOIN减少数据库往返
    2. 全文搜索索引提高搜索性能
    3. 分页查询控制内存使用
    4. 预计算字段减少实时计算
    
    查询构建过程:
    1. 构建基础WHERE条件
    2. 添加动态筛选条件
    3. 应用全文搜索
    4. 设置排序和分页
    5. 执行查询并处理结果
    """
    # 构建基础查询条件
    where_conditions = [
        "sl.status = 'APPROVED'",
        "sl.is_deleted = false",
    ]
    
    query_params = []
    
    # 动态添加筛选条件
    if featured:
        where_conditions.append("sl.is_featured = true")
    
    if creators:
        placeholders = ",".join(["%s"] * len(creators))
        where_conditions.append(f"up.username IN ({placeholders})")
        query_params.extend(creators)
    
    if search_query:
        # 全文搜索支持
        where_conditions.append("""
            (
                sl.name ILIKE %s OR 
                sl.description ILIKE %s OR
                to_tsvector('english', sl.name || ' ' || sl.description) 
                @@ plainto_tsquery('english', %s)
            )
        """)
        search_pattern = f"%{search_query}%"
        query_params.extend([search_pattern, search_pattern, search_query])
    
    # 排序策略
    order_by = "sl.created_at DESC"
    if sorted_by == "runs":
        order_by = "sl.runs DESC, sl.rating DESC"
    elif sorted_by == "rating":
        order_by = "sl.rating DESC, sl.review_count DESC"
    
    # 构建完整查询
    base_query = f"""
        SELECT 
            sl.*,
            up.username as creator_username,
            up.name as creator_name,
            COUNT(*) OVER() as total_count
        FROM store_listings sl
        JOIN user_profiles up ON sl.user_id = up.user_id
        WHERE {' AND '.join(where_conditions)}
        ORDER BY {order_by}
        LIMIT %s OFFSET %s
    """
    
    offset = (page - 1) * page_size
    query_params.extend([page_size, offset])
    
    # 执行查询
    results = await database.fetch_all(base_query, query_params)
    
    # 处理结果
    agents = []
    total_count = 0
    
    for row in results:
        if total_count == 0:
            total_count = row["total_count"]
        
        agent = StoreAgent(
            id=row["id"],
            name=row["name"],
            description=row["description"],
            category=row["category"],
            rating=row["rating"],
            runs=row["runs"],
            creator=StoreCreator(
                username=row["creator_username"],
                name=row["creator_name"],
            ),
            created_at=row["created_at"],
        )
        agents.append(agent)
    
    return StoreAgentsResponse(
        agents=agents,
        total_count=total_count,
        page=page,
        page_size=page_size,
        total_pages=(total_count + page_size - 1) // page_size,
    )

# /autogpt_platform/backend/backend/server/rest_api.py

class AgentServer(backend.util.service.AppProcess):
    """
    Agent服务器进程类
    
    继承自AppProcess基类，实现统一的服务生命周期管理
    """
    
    def run(self):
        """
        启动FastAPI服务器
        
        配置说明:
        1. 从环境配置读取服务器参数
        2. 设置日志级别和格式
        3. 配置热重载（开发环境）
        4. 启动uvicorn服务器
        """
        server_app = uvicorn.Server(
            uvicorn.Config(
                app,
                host=settings.config.backend_host,
                port=settings.config.backend_port,
                log_level="info",
                reload=settings.config.backend_reload,
                access_log=True,
            )
        )
        
        with contextlib.suppress(KeyboardInterrupt):
            server_app.run()

# /autogpt_platform/backend/backend/app.py

def main(**kwargs):
    """
    AutoGPT服务器主入口函数
    
    启动组件顺序:
    1. DatabaseManager: 数据库连接管理
    2. Scheduler: 定时任务调度器
    3. NotificationManager: 通知服务
    4. WebsocketServer: WebSocket服务器
    5. AgentServer: REST API服务器
    6. ExecutionManager: 执行引擎管理器
    
    组件启动顺序很重要，数据库服务必须最先启动，
    其他服务按依赖关系顺序启动
    """
    from backend.executor import DatabaseManager, ExecutionManager, Scheduler
    from backend.notifications import NotificationManager
    from backend.server.rest_api import AgentServer
    from backend.server.ws_api import WebsocketServer
    
    run_processes(
        DatabaseManager(),
        Scheduler(),
        NotificationManager(),
        WebsocketServer(),
        AgentServer(),
        ExecutionManager(),
        **kwargs,
    )

def run_processes(*processes: "AppProcess", **kwargs):
    """
    运行多个后台进程
    
    功能说明:
    1. 注册系统信号处理器
    2. 按顺序启动所有服务进程
    3. 等待停止信号
    4. 优雅停机，确保资源正确释放
    
    参数:
        processes: AppProcess实例列表
        kwargs: 传递给各进程的参数
    """
    stop_event = threading.Event()
    
    def signal_handler(signum, frame):
        stop_event.set()
    
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    
    # 启动所有进程
    for process in processes:
        process.start(**kwargs)
    
    # 等待停止信号
    stop_event.wait()
    
    # 优雅停机
    for process in processes:
        process.stop()