| 12345678910111213141516171819202122232425262728293031 |
- """Agent Runtime Event 与 SSE 传输协议。"""
- import json
- from datetime import datetime
- from typing import Any, Literal
- from pydantic import BaseModel, ConfigDict, Field
- class RuntimeEvent(BaseModel):
- model_config = ConfigDict(extra="forbid")
- event: Literal[
- "run.started",
- "tool.completed",
- "ui.ready",
- "run.completed",
- "run.failed",
- ]
- run_id: str
- sequence: int = Field(ge=1)
- occurred_at: datetime
- data: dict[str, Any] = Field(default_factory=dict)
- def to_sse(self) -> str:
- payload = self.model_dump(mode="json")
- return (
- f"id: {self.run_id}:{self.sequence}\n"
- f"event: {self.event}\n"
- f"data: {json.dumps(payload, ensure_ascii=False, separators=(',', ':'))}\n\n"
- )
|
