sdk应用框架-j9九游会登录

框架概述

sdk应用框架基于starlette web框架实现，采用装饰器模式将用户函数封装为http端点。这种设计使得开发者无需关心底层网络通信，只需专注于业务逻辑的实现。

核心入口类: agentartsruntimeapp

agentartsruntimeapp是sdk的核心类，继承自starlette框架，提供ai代理部署所需的所有web服务功能。

文件位置：src/agentarts/sdk/runtime/app.py

基本用法：

from agentarts.sdk import agentartsruntimeapp, requestcontext
app = agentartsruntimeapp()
@app.entrypoint
async def handler(payload: dict, context: requestcontext = none) -> dict:
    """agent 入口函数"""
    message = payload.get("message", "")
    return {"response": f"received: {message}"}
if __name__ == "__main__":
    app.run()

装饰器详解

@app.entrypoint - 主入口装饰器

用于注册主处理函数，处理所有到达/invocations端点的请求。

@app.entrypoint
def handle_request(payload, context):
    """
    处理agent调用请求
    args:
        payload: 请求载荷，包含用户输入
        context: 请求上下文，包含用户id、会话id等信息
    returns:
        响应数据，可以是普通对象、生成器或异步生成器
    """
    prompt = payload.get("prompt")
    # 处理逻辑
    return {"response": f"处理: {prompt}"}

支持的返回类型：

• 普通对象：自动序列化为json响应
• 同步生成器：支持流式响应（sse）
• 异步生成器：支持异步流式响应

@app.ping - 健康检查装饰器

用于注册自定义的健康检查处理器。

@app.ping
def custom_ping():
    """返回自定义的ping状态"""
    return "healthy"  # 或返回 pingstatus.healthy

pingstatus枚举值：

• healthy：健康状态
• healthy_busy：忙碌状态
• unhealthy：不健康状态

@app.websocket - websocket装饰器

用于注册websocket处理函数，支持双向实时通信。

@app.websocket
async def handle_websocket(websocket, context):
    """处理websocket连接"""
    await websocket.accept()
    while true:
        try:
            data = await websocket.receive_text()
            # 处理消息
            await websocket.send_text(f"echo: {data}")
        except exception:
            break

@app.async_task - 异步任务装饰器

用于注册异步任务，sdk会自动追踪任务健康状态。

@app.async_task
async def background_task():
    """后台异步任务"""
    # 任务执行期间，ping状态会自动设置为healthy_busy
    # 任务完成后，自动恢复为healthy
    await asyncio.sleep(10)

请求上下文

sdk提供两种上下文对象用于获取请求信息：

agentartsruntimecontext

基于contextvars的线程安全运行时上下文，用于在请求处理过程中存储和获取上下文信息。

from agentarts.sdk.runtime.context import agentartsruntimecontext
# 设置上下文
agentartsruntimecontext.set_user_id("user123")
agentartsruntimecontext.set_session_id("session456")
agentartsruntimecontext.set_workload_access_token("token789")
# 获取上下文
user_id = agentartsruntimecontext.get_user_id()
session_id = agentartsruntimecontext.get_session_id()
token = agentartsruntimecontext.get_workload_access_token()

requestcontext

请求级别的上下文对象，包含请求的详细信息。

@app.entrypoint
def handle_request(payload, context: requestcontext):
    """
    context包含以下属性：
    - user_id: 用户id
    - session_id: 会话id
    - request_id: 请求id
    - workload_access_token: 工作负载访问令牌
    """
    print(f"user: {context.user_id}")
    print(f"session: {context.session_id}")
    return {"status": "ok"}

框架请求处理流程

http请求
    │
    ▼
starlette router
    │
    ├─ post /invocations ──▶ _handle_invocation()
    │                              │
    │                              ├─ 解析json payload
    │                              ├─ 构建requestcontext
    │                              ├─ 调用用户handler
    │                              ├─ 处理响应类型:
    │                              │   ├─ streamingresponse (sse)
    │                              │   ├─ jsonresponse
    │                              │   └─ 普通response
    │                              └─ 返回响应
    │
    ├─ get /ping ──▶ _handle_ping()
    │                     │
    │                     └─ 返回pingstatus
    │
    └─ websocket /ws ──▶ _handle_websocket()
                              │
                              └─ 传递给websocket handler