Agent Runtime：线程隔离、SSE、取消和恢复

{"event":"tool_started","run_id":"run-1","thread_id":"task-9","seq":2,"data":{"tool":"get_status"}}

python -m pip install "fastapi>=0.115,<1" "uvicorn>=0.30,<1"

import asyncio
import json
import uuid
from dataclasses import dataclass, field

from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse


app = FastAPI()


@dataclass
class Task:
    owner: str
    run_id: str
    queue: asyncio.Queue = field(default_factory=asyncio.Queue)
    cancel: asyncio.Event = field(default_factory=asyncio.Event)
    status: str = "running"


tasks: dict[str, Task] = {}


def owned_task(thread_id: str, user_id: str) -> Task:
    task = tasks.get(thread_id)
    if not task or task.owner != user_id:
        raise HTTPException(404, "task not found")
    return task


async def execute(thread_id: str, task: Task) -> None:
    for seq, step in enumerate(("plan", "get_status", "summarize"), start=1):
        if task.cancel.is_set():
            task.status = "cancelled"
            await task.queue.put({"event": "cancelled", "seq": seq})
            break
        await task.queue.put({"event": "step_started", "seq": seq, "data": {"step": step}})
        await asyncio.sleep(0.2)  # 替换为支持超时的真实节点
    else:
        task.status = "completed"
        await task.queue.put({"event": "completed", "seq": 4})
    await task.queue.put(None)


@app.post("/tasks/{thread_id}")
async def start(thread_id: str, x_user_id: str = Header()) -> dict:
    if thread_id in tasks:
        raise HTTPException(409, "thread already exists")
    task = Task(owner=x_user_id, run_id=str(uuid.uuid4()))
    tasks[thread_id] = task
    asyncio.create_task(execute(thread_id, task))
    return {"thread_id": thread_id, "run_id": task.run_id}


@app.get("/tasks/{thread_id}/events")
async def events(thread_id: str, x_user_id: str = Header()):
    task = owned_task(thread_id, x_user_id)

    async def stream():
        while True:
            item = await task.queue.get()
            if item is None:
                break
            item.update({"thread_id": thread_id, "run_id": task.run_id})
            yield f"data: {json.dumps(item, ensure_ascii=False)}\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")


@app.post("/tasks/{thread_id}/cancel")
async def cancel(thread_id: str, x_user_id: str = Header()) -> dict:
    task = owned_task(thread_id, x_user_id)
    task.cancel.set()
    return {"status": "cancelling"}