diff --git a/a2a-t-sample/README-zh.md b/a2a-t-sample/README-zh.md index f9d5d495..ee7f452e 100644 --- a/a2a-t-sample/README-zh.md +++ b/a2a-t-sample/README-zh.md @@ -13,4 +13,6 @@ | 目录 | 说明 | | --- | --- | | [fault-management/](fault-management/) | 事件订阅样例——流式推送 Incident 通知 | +| [negotiation/](negotiation/) | 协商闭环样例——离线 propose → accept 往返 | +| [observability-sample/](observability-sample/) | 可观测样例——所有场景的 trace / 指标 / 日志，可选 Langfuse 后端 | | [ran-energy-saving/](ran-energy-saving/) | 无线网络节能样例——运行 RAN 节能任务，并包含三类 Negotiation-T 协商（信息 / 目标 / 可行性） | diff --git a/a2a-t-sample/README.md b/a2a-t-sample/README.md index 0e975e08..301c35f4 100644 --- a/a2a-t-sample/README.md +++ b/a2a-t-sample/README.md @@ -13,4 +13,6 @@ See the repository root [`README.md`](../README.md) for setup and how to run. | Directory | Description | | --- | --- | | [fault-management/](fault-management/) | Event subscription sample — stream Incident notifications | +| [negotiation/](negotiation/) | Negotiation closed-loop sample — offline propose → accept round trips | +| [observability-sample/](observability-sample/) | Observability sample — traces, metrics and logs for every scenario, with an optional Langfuse backend | | [ran-energy-saving/](ran-energy-saving/) | RAN energy saving sample — run a RAN energy-saving task, including Negotiation-T negotiations in three kinds (information / target / feasibility) | diff --git a/a2a-t-sample/observability-sample/README-zh.md b/a2a-t-sample/observability-sample/README-zh.md new file mode 100644 index 00000000..299263b3 --- /dev/null +++ b/a2a-t-sample/observability-sample/README-zh.md @@ -0,0 +1,133 @@ +# observability-sample — 看见你的 A2A-T 智能体在做什么 + +一个可直接运行的样例：展示 **A2A-T 交互的完整生命周期**——客户端发起请求、服务端处理、 +结果流式返回——每一步都由 `a2a-t-sdk` 的可观测层自动记录（trace / 指标 / 日志）。 + +样例里**没有任何手写的 OpenTelemetry 代码**：服务端加一个包装器、客户端做一次注册， +其余全部由 SDK 自动完成。 + +--- + +## 可以体验什么 + +| 场景（`--scenario=`） | 故事 | +|------------------------|------| +| `native` | 一个任务完整走完：客户端提交、服务端处理后流式返回 artifact。观察请求如何把两端"点亮"。 | +| `a2at` | 用 LLM 生成任务 prompt（默认 mock LLM——无需 API Key），并在 SDK 调用外手动加一个"L4" Span。 | +| `negotiation` | 智能体**先谈价、后干活**：两轮报价通过纯消息完成（此时还没有任务），双方达成一致后，真正的任务才启动并流式返回结果。 | + +## 快速开始 + +前置条件：已在仓库根目录，且安装了 [uv](https://docs.astral.sh/uv/)。 + +```bash +# 仓库根目录——终端 1：启动服务端（端口 8100） +cd a2a-t-sample +export PYTHONPATH="$PWD/observability-sample/src" +export A2AT_OBSERVABILITY_ENABLED=true +uv run python -m obs_sample.server_main + +# 终端 2：运行客户端场景 +cd a2a-t-sample +export PYTHONPATH="$PWD/observability-sample/src" +uv run python -m obs_sample.client_main --scenario=native +``` + +Windows PowerShell：把 `export` 换成 `$env:PYTHONPATH = "$PWD\observability-sample\src"`。 + +两个终端都会输出日志（`[server] ...` / `[client] ...`）；默认配置下服务端还会把每个 +span 以 JSON 打印到控制台。 + +## 场景说明 + +### native — 一个任务，两端可见 + +1. 终端 1 启动服务端；终端 2 以 `--scenario=native` 运行客户端。 +2. 客户端提交任务；服务端经历 *submitted → working* 并流式返回 artifact。 +3. 每一步在**两端**都以 span 呈现，并汇入同一条 trace（客户端自动把 W3C `traceparent` + 传给服务端）。 + +### a2at — 用 LLM 生成 prompt + +```bash +uv run python -m obs_sample.client_main --scenario=a2at +``` + +客户端通过 SDK 的 LLM 门面生成任务 prompt。`a2a-t-sample/.env` 里 `A2AT_LLM_API_KEY` +留空时会使用**mock LLM**（固定回复），完全离线可跑。LLM 的 token 用量会被自动计量并 +挂到 trace 上。 + +### negotiation — 先谈成，再执行 + +```bash +# 终端 1：以协商模式重启服务端 +uv run python -m obs_sample.server_main --scenario=negotiation +# 终端 2： +uv run python -m obs_sample.client_main --scenario=negotiation +``` + +这是 Negotiation-T 谈判的真实时序： + +1. **第 1 轮** — 客户端出价；服务端还一个不同的价并断开连接。*此时还没有任务—— + 谈判通过纯消息进行。* +2. **第 2 轮** — 客户端重连并接受对方的还价；服务端确认达成一致，再次断开。 +3. **达成一致后任务才启动** — 客户端以独立请求启动真正的任务生命周期并接收结果。 + +每一轮都是一次独立交互；样例把它们串起来，让你能在同一个地方看到整场谈判 +（见下方 Langfuse）。 + +## 把所有数据发到 Langfuse（可选） + +在启动**两个进程之前**设置三个变量，所有 span 就会进入你的 Langfuse 项目—— +每次交互一条可浏览的 trace： + +```bash +export LANGFUSE_PUBLIC_KEY="pk-lf-..." +export LANGFUSE_SECRET_KEY="sk-lf-..." +export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com" +``` + +在 Langfuse 界面里，每一轮的客户端/服务端 span 和最终任务会呈现在同一条时间线上—— +谁出了什么价、哪一轮被接受、每一步耗时多久，一目了然。 + +## 记录了什么（速查表） + +| 你做的事 | SDK 记录的内容 | +|----------|----------------| +| 客户端发请求 | `SendStreamingMessage` span（CLIENT）+ 耗时指标 | +| 服务端收到请求 | `SendStreamingMessage` span（SERVER），是客户端 span 的子节点 | +| 服务端流式发出事件 | `SendStreamingMessage-event` span，LINK 关联入口 | +| 发出 / 收到协商消息 | `SendStreamingMessage-negotiation` span（含轮次与施为动词） | +| 服务端推送通知 | `SendMessage-pushNotification` span，LINK 关联服务端入口 | +| LLM 消耗 token | token 用量指标 + 下一跳 trace 上的 `gen_ai.usage.*` | +| 值得记录的事件 | logger `a2at.observability` 上的结构化日志 | + +三类信号可独立开关——见下方环境变量。 + +## 环境变量 + +常用： + +| 变量 | 默认值 | 含义 | +|------|--------|------| +| `A2AT_OBSERVABILITY_ENABLED` | `true` | 总开关——`false` 时 SDK 完全透明 | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | 未设置 | 把 trace/指标/日志发到 OTLP collector，而不是控制台 | +| `A2AT_CONSOLE_EXPORTER` | `true` | `false` 关闭 dev 模式控制台兜底（stdout 不再输出 JSON） | +| `OTEL_SERVICE_NAME`（或 `A2AT_SERVICE_NAME`） | `a2a-t-agent` | 本进程在后端的名称 | +| `A2AT_OBS_SAMPLE_PORT` | `8100` | 样例服务端端口 | +| `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_BASE_URL` | 未设置 | 启用 Langfuse 后端 | + +进阶（较少用到）：`A2AT_TRACE_ENABLED`、`A2AT_METRIC_ENABLED`、`A2AT_LOG_ENABLED` +（按信号独立开关——例如关 trace 后指标与日志仍工作）、`A2AT_EXPORTER_PROTOCOL` +（`grpc` / `http/protobuf`）、`OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED=false`（硬关闭）。 + +## 说明 + +- 服务端使用**官方默认**的 `DefaultRequestHandler`（a2a-sdk 1.1.x 的别名，指向 + `DefaultRequestHandlerV2`）：可观测装饰器会自动检测 V2 形状，并用装饰后的 + executor / push sender 重建其任务注册表——per-event 与 push span 在官方 handler + 上同样生效。（`LegacyRequestHandler` 通过逐属性包装的方式继续受支持。） +- 推送通知会回环到服务端的 `/push-sink`，因此样例能端到端展示 + `SendMessage-pushNotification` span。 +- 离线验证辅助脚本与样例代码放在一起（`check_*.py`、`verify_langfuse.py`）——它们是 + 一次性调试工具，不属于样例主流程。 diff --git a/a2a-t-sample/observability-sample/README.md b/a2a-t-sample/observability-sample/README.md index 826e83b2..87a20497 100644 --- a/a2a-t-sample/observability-sample/README.md +++ b/a2a-t-sample/observability-sample/README.md @@ -1,98 +1,139 @@ -# observability-sample — A2A-T 可观测性演示（v3.0 装饰器架构） +# observability-sample — see what your A2A-T agents are actually doing -基于 `a2a-t-sdk[observability]` v3.0 的端到端可观测性样例：一行装饰器接线、零手动 OTel 代码。 +A runnable sample that shows the **full lifecycle of an A2A-T interaction** — a client +asks, a server answers, artifacts stream back — with every step automatically traced, +measured and logged by the `a2a-t-sdk` observability layer. -## 接线（v3.0） +There is **no manual OpenTelemetry code** in the sample: the server adds one wrapper, +the client adds one registration, and everything else is emitted by the SDK. -- Server：`A2ATRequestHandlerDecorator(LegacyRequestHandler(...))` —— 自动创建 SERVER 入口 Span - （从 `ServerCallContext.state["headers"]` 提取 traceparent 作为父上下文，无需 ASGI 中间件）、 - 包装 executor 产生 per-event Span、包装 push sender 产生 push Span -- Client：`register_client_factory(factory)` —— 之后 `factory.create(card)` 产出的 client 自动携带 - 观测 transport（CLIENT Span + traceparent 注入，经 `ClientCallContext.service_parameters` 落到 HTTP 头） -- OTel 自动配置：SDK 在首次使用时自动装配 Tracer/Meter/LoggerProvider（未配置 - `OTEL_EXPORTER_OTLP_ENDPOINT` 时回退 Console 导出），样例不含任何手动 setup 代码 +--- -## a2a-sdk 1.1.x handler caveat（必须 LegacyRequestHandler） +## What you can try -完整 per-event Span 富化需要 `LegacyRequestHandler`。a2a-sdk 1.1.2 默认的 `DefaultRequestHandler` -（V2）在构造期即捕获裸的 `agent_executor`/`_push_sender` 引用（ActiveTaskRegistry），装饰器构造期的 -executor/push 包装会被绕过。本样例固定使用 `LegacyRequestHandler`（服务端行为与 V2 兼容）。 +| Scenario (`--scenario=`) | Story | +|--------------------------|-------| +| `native` | A task flows end to end: the client submits, the server works and streams artifacts back. Watch the request light up on both sides. | +| `a2at` | Prompt generation with an LLM (a mock LLM by default — no API key needed) and a manual "L4" span around the SDK call. | +| `negotiation` | The agent **bargains before it commits**: two offer rounds happen over plain messages (no task yet), and only after both sides agree does the real task start and stream its result. | -## 运行（离线，默认 Console 输出） +## Quick start -```powershell -cd a2a-t-sample -$env:PYTHONPATH = "$pwd\observability-sample\src" -$env:A2AT_OBSERVABILITY_ENABLED = "true" # 观测总开关（默认即 true；置 false 时装饰器完全直通） +Prerequisites: you are in the repository root and [uv](https://docs.astral.sh/uv/) is installed. -# 终端 1：server（端口 8100） +```bash +# from the repo root — terminal 1: start the server (port 8100) +cd a2a-t-sample +export PYTHONPATH="$PWD/observability-sample/src" +export A2AT_OBSERVABILITY_ENABLED=true uv run python -m obs_sample.server_main -# 终端 2：client（原生 a2a-python 流量） +# terminal 2: run a client scenario +cd a2a-t-sample +export PYTHONPATH="$PWD/observability-sample/src" uv run python -m obs_sample.client_main --scenario=native - -# 或 client（A2ATClient 模板生成 + trace_facade L4 Span，需仓库 a2a-t-sample/.env： -# 不存在时从 env.example 复制并将 A2AT_LLM_API_KEY 置空（A2AT_LLM_API_KEY=），置空后自动走 mock LLM） -uv run python -m obs_sample.client_main --scenario=a2at - -# 协商场景：server 需以 negotiation 模式启动（返回 Negotiation-T 终态 Message） -uv run python -m obs_sample.server_main --scenario=negotiation -# 然后另一个终端： -uv run python -m obs_sample.client_main --scenario=negotiation ``` -`--scenario=a2at` 分支会调用一次幂等的 `setup()`（SDK 公共自动配置入口）：L4 门面 Span -产生在首个 A2A 请求之前，需先完成 OTel 自动装配才不会丢失；native 场景无需任何调用， -首个请求即触发自动配置。 +Windows PowerShell: use `$env:PYTHONPATH = "$PWD\observability-sample\src"` instead of `export`. -端口可用 `A2AT_OBS_SAMPLE_PORT` 覆盖（默认 8100）。样例 client 会请求 push notification -（回环到 server 的 `/push-sink`），用于演示 `SendMessage-pushNotification` Span。 +You will see log lines on both terminals (`[server] ...` / `[client] ...`), and — with the +default configuration — every span printed as JSON on the server console. -## 预期输出（v3.0 Span 命名，a2a-java 对齐） +## Scenario walkthroughs -Client 终端（CLIENT Span）： +### native — one task, both sides -- `SendStreamingMessage`（入口 Span，含 `extension.name`、`gen_ai.operation.name`、 - `gen_ai.client.operation.duration` 指标；traceparent 注入请求头） -- `SendStreamingMessage-event`（per-event，LINK 到入口 Span） -- `--scenario=a2at` 另有 `a2at.sdk.client.generate_task_prompt`（L4） -- `--scenario=negotiation` 另有 `SendStreamingMessage-negotiation`（PARENT 到入口 Span， - 含 `negotiation.id/round/max_rounds/performative` 属性；服务端返回 Negotiation-T 终态 Message 时自动识别） +1. Terminal 1 starts the server. Terminal 2 runs the client with `--scenario=native`. +2. The client submits a task; the server moves it through *submitted → working* and + streams an artifact back. +3. Each step appears as a span on **both** sides, joined into one trace + (the client passes a W3C `traceparent` to the server automatically). -Server 终端（SERVER/INTERNAL Span，与 Client 同 trace_id —— W3C traceparent 端到端贯通）： +### a2at — generate a prompt with an LLM -- `SendStreamingMessage`（SERVER 入口 Span，父 = Client 入口 Span） -- `SendStreamingMessage-event`（per-event INTERNAL Span，父 = SERVER 入口 Span） -- `SendMessage-pushNotification`（CLIENT kind Span，LINK 到 SERVER 入口 Span） -- `--scenario=negotiation` 时 server 返回终态协商 Message（不产生 per-event Span 流） +```bash +uv run python -m obs_sample.client_main --scenario=a2at +``` -日志与指标： +The client generates a task prompt through the SDK's LLM facade. With an empty +`A2AT_LLM_API_KEY` in `a2a-t-sample/.env` a canned **mock LLM** answers, so this runs +fully offline. The LLM's token usage is measured automatically and attached to the trace. -- logger `a2at.observability` 自动经 OTLP/Console 导出（`task.status_changed` / `task.artifact` 等） -- 指标：`gen_ai.client.operation.duration`（client）、`a2at.task.request.duration`（side=client/server） +### negotiation — bargain first, execute later + +```bash +# terminal 1: restart the server in negotiation mode +uv run python -m obs_sample.server_main --scenario=negotiation +# terminal 2: +uv run python -m obs_sample.client_main --scenario=negotiation +``` -## 配置开关 +This is the real-world timing of a Negotiation-T deal: -| 环境变量 | 默认 | 说明 | -| --- | --- | --- | -| `A2AT_OBSERVABILITY_ENABLED` | `true` | 观测总开关；`false` → 装饰器完全直通（无任何观测） | -| `A2AT_TRACE_ENABLED` | `true` | Trace 信号开关；`false` → 不产生 Span，指标/日志仍工作 | -| `A2AT_METRIC_ENABLED` | `true` | Metric 信号开关 | -| `A2AT_LOG_ENABLED` | `true` | Log 信号开关 | -| `OTEL_EXPORTER_OTLP_ENDPOINT` | 未设置 | 未设置 → Console 导出（dev 模式） | -| `A2AT_EXPORTER_PROTOCOL` | `grpc` | OTLP 协议（`grpc` / `http/protobuf`） | -| `OTEL_SERVICE_NAME`（或 `A2AT_SERVICE_NAME`） | `a2a-t-agent` | Resource `service.name` | -| `OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED` | `true` | OTel 主开关；`false` → 全部 NoOp 直通 | +1. **Round 1** — the client proposes; the server counters with a different offer and + disconnects. *No task exists yet — bargaining happens over plain messages.* +2. **Round 2** — the client reconnects, accepts the counter-offer; the server confirms + the agreement and disconnects again. +3. **The task starts only after agreement** — a separate request now runs the full + task lifecycle and streams the result. -## 切换 OTLP 后端 +Each round is an independent exchange; the sample ties them together so you can follow +the whole deal in one place (see Langfuse below). -```powershell -$env:OTEL_EXPORTER_OTLP_ENDPOINT = "http://collector:4317" -``` +## Send everything to Langfuse (optional) -（需另装 `opentelemetry-exporter-otlp-proto-grpc`；默认 Console 输出无需任何额外依赖。） +Set three variables before starting **both** processes and every span lands in your +Langfuse project as one browsable trace per interaction: -## 说明 +```bash +export LANGFUSE_PUBLIC_KEY="pk-lf-..." +export LANGFUSE_SECRET_KEY="sk-lf-..." +export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com" +``` -- 断言式端到端验证见 `tests/integration/test_e2e_observability.py`（`A2AT_TEST_A2A=1`）。 -- a2at 场景的 mock LLM 资源复用 `../subscribe-incident/resources/mock_responses`。 +In the Langfuse UI you will see the client and server spans of every round and the final +task on a single timeline — including who proposed what, which round was accepted, and +how long each step took. + +## What gets recorded (cheat sheet) + +| You do | The SDK records | +|--------|-----------------| +| Client sends a request | `SendStreamingMessage` span (CLIENT) + duration metric | +| Server receives it | `SendStreamingMessage` span (SERVER), child of the client's | +| Server streams an event | `SendStreamingMessage-event` span, linked to the entry | +| A negotiation message is sent / received | `SendStreamingMessage-negotiation` span with round & performative | +| Server pushes a notification | `SendMessage-pushNotification` span, linked to the server entry | +| The LLM generates tokens | token-usage metric + `gen_ai.usage.*` on the next trace | +| Anything noteworthy | a structured log line on logger `a2at.observability` | + +Signals can be toggled independently — see the environment variables below. + +## Environment variables + +Commonly used: + +| Variable | Default | Meaning | +|----------|---------|---------| +| `A2AT_OBSERVABILITY_ENABLED` | `true` | Master switch — `false` makes the SDK fully transparent | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | *(unset)* | Send spans/metrics/logs to an OTLP collector instead of the console | +| `A2AT_CONSOLE_EXPORTER` | `true` | `false` disables the dev-mode console fallback (no stdout JSON) | +| `OTEL_SERVICE_NAME` (or `A2AT_SERVICE_NAME`) | `a2a-t-agent` | How this process is named in the backend | +| `A2AT_OBS_SAMPLE_PORT` | `8100` | Sample server port | +| `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_BASE_URL` | *(unset)* | Enable the Langfuse backend | + +Advanced (rarely needed): `A2AT_TRACE_ENABLED`, `A2AT_METRIC_ENABLED`, `A2AT_LOG_ENABLED` +(per-signal switches — e.g. trace off keeps metrics and logs alive), `A2AT_EXPORTER_PROTOCOL` +(`grpc` / `http/protobuf`), `OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED=false` (hard disable). + +## Notes + +- The server uses the **official default** `DefaultRequestHandler` (a2a-sdk 1.1.x alias + for `DefaultRequestHandlerV2`): the observability decorator detects the V2 shape and + rebuilds its task registry with the decorated executor / push sender, so per-event and + push spans work on the official handler too. (`LegacyRequestHandler` remains supported + via per-attribute wrapping.) +- Push notifications are echoed back to the server's `/push-sink` endpoint, which is + why the sample can show the `SendMessage-pushNotification` span end to end. +- Offline verification helpers live next to the sample code (`check_*.py`, + `verify_langfuse.py`) — they are one-off debugging tools, not part of the sample flow. diff --git a/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py b/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py index eb89ef63..84f9f81b 100644 --- a/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py +++ b/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py @@ -6,7 +6,7 @@ responses (subscribe-incident does the latter in its llm_logger, which this sample does not carry). -Mock response files live under ../subscribe-incident/resources/mock_responses/. +Mock response files live under ../fault-management/resources/mock_responses/. """ from __future__ import annotations @@ -19,8 +19,14 @@ from dotenv import dotenv_values # mock_llm.py → obs_sample → src → observability-sample → a2a-t-sample (parents[3]), -# then into subscribe-incident's shared mock resources. -_RESOURCES_DIR = Path(__file__).resolve().parents[3] / "subscribe-incident" / "resources" / "mock_responses" +# then into the shared mock resources (the directory was renamed from +# subscribe-incident to fault-management in #74; both names are accepted so the +# sample keeps working on older checkouts). +_MOCK_DIR_CANDIDATES = [ + Path(__file__).resolve().parents[3] / "fault-management" / "resources" / "mock_responses", + Path(__file__).resolve().parents[3] / "subscribe-incident" / "resources" / "mock_responses", +] +_RESOURCES_DIR = next((p for p in _MOCK_DIR_CANDIDATES if p.exists()), _MOCK_DIR_CANDIDATES[0]) _MOCK_RESPONSE_FILES = [ "scenario_recognition.json", diff --git a/a2a-t-sample/observability-sample/src/obs_sample/server_main.py b/a2a-t-sample/observability-sample/src/obs_sample/server_main.py index cde982e3..bed082e2 100644 --- a/a2a-t-sample/observability-sample/src/obs_sample/server_main.py +++ b/a2a-t-sample/observability-sample/src/obs_sample/server_main.py @@ -1,14 +1,12 @@ """Observability sample server (v3.0 decorator architecture, no LLM needed). -One-line wiring: ``A2ATRequestHandlerDecorator(LegacyRequestHandler(...))`` creates the -SERVER entry spans, wraps the executor for per-event spans and wraps the push sender — -no ASGI middleware and no manual OTel setup (the SDK auto-configures Console/OTLP +One-line wiring: ``A2ATRequestHandlerDecorator(DefaultRequestHandler(...))`` creates the +SERVER entry spans, wraps the executor for per-event spans and wraps the push sender —?no ASGI middleware and no manual OTel setup (the SDK auto-configures Console/OTLP exporters on first use when ``A2AT_OBSERVABILITY_ENABLED=true``). -Uses ``LegacyRequestHandler``: a2a-sdk 1.1.2's default ``DefaultRequestHandler`` (V2) -captures raw ``agent_executor``/``_push_sender`` references at construction time, so the -decorator's post-construction wraps would be bypassed (see README "a2a-sdk 1.1.x handler -caveat"). +Uses the official default ``DefaultRequestHandler`` (a2a-sdk 1.1.x): the decorator +detects the V2 shape and rebuilds its ``ActiveTaskRegistry`` with the decorated +executor / push sender, so per-event and push spans work on the official handler too. """ from __future__ import annotations @@ -22,12 +20,19 @@ from a2a.server.agent_execution.agent_executor import AgentExecutor from a2a.server.agent_execution.context import RequestContext from a2a.server.events.event_queue import EventQueue -from a2a.server.request_handlers.default_request_handler import LegacyRequestHandler +from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.routes import create_agent_card_routes, create_rest_routes from a2a.server.tasks.base_push_notification_sender import BasePushNotificationSender from a2a.server.tasks.inmemory_push_notification_config_store import InMemoryPushNotificationConfigStore from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore -from a2a.types import Artifact, TaskArtifactUpdateEvent, TaskState, TaskStatus, TaskStatusUpdateEvent +from a2a.types import ( + Artifact, + Task, + TaskArtifactUpdateEvent, + TaskState, + TaskStatus, + TaskStatusUpdateEvent, +) from a2a.utils.constants import TransportProtocol from a2a_t.observability import A2ATRequestHandlerDecorator from google.protobuf.json_format import MessageToDict, ParseDict @@ -43,11 +48,19 @@ def _sample_port() -> int: class EchoExecutor(AgentExecutor): - """Emits submitted -> working -> artifact -> completed; mirrors a Task-T streaming flow.""" + """Emits the initial Task snapshot -> submitted -> working -> artifact -> completed + (V2 protocol: the Task snapshot must precede any status event).""" async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: task_id = context.task_id or "T-sample" context_id = context.context_id or "C-sample" + await event_queue.enqueue_event( + Task( + id=task_id, + context_id=context_id, + status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED), + ) + ) for state in (TaskState.TASK_STATE_SUBMITTED, TaskState.TASK_STATE_WORKING): await event_queue.enqueue_event( TaskStatusUpdateEvent(task_id=task_id, context_id=context_id, status=TaskStatus(state=state)) @@ -74,11 +87,11 @@ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None class NegotiationExecutor(AgentExecutor): """Negotiation-T flow with real per-round timing (2026-09-30): - - **Negotiation rounds happen over bare Messages — NO task is created before + - **Negotiation rounds happen over bare Messages —?NO task is created before agreement.** Each round: the client sends a Message carrying the Negotiation-T metadata + ``negotiationContext``; the server responds with a single negotiation Message (counter-offer, or ACCEPT confirmation for the - accepting round) and the stream ends — the server disconnects after the + accepting round) and the stream ends —?the server disconnects after the Message. No task lifecycle events during rounds. - **The task starts only AFTER agreement**: a separate task request (Task-T metadata) triggers the full task lifecycle (submitted -> working -> @@ -87,7 +100,7 @@ class NegotiationExecutor(AgentExecutor): The decorators create the ``SendStreamingMessage-negotiation`` spans (PARENT -> entry, both ends) during rounds, and the regular per-event spans only for the post-agreement task request. Each round is an independent - trace; backends aggregate rounds by ``negotiation.id`` (spec §6.3 mode A). + trace; backends aggregate rounds by ``negotiation.id`` (spec 搂6.3 mode A). """ async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: @@ -102,7 +115,7 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non async def _negotiation_round( self, context: RequestContext, event_queue: EventQueue, negotiation_context: dict[str, Any] ) -> None: - """One negotiation round: respond with a single negotiation Message — no task.""" + """One negotiation round: respond with a single negotiation Message —?no task.""" incoming_round = int(negotiation_context.get("round", 1)) performative = str(negotiation_context.get("performative", "PROPOSE")) if performative == "PROPOSE": @@ -129,7 +142,7 @@ async def _negotiation_round( message.metadata[_NEGOTIATION_T_URI] = offer message.metadata["negotiationContext"] = response_context await event_queue.enqueue_event(message) - # Stream ends here on purpose: no task events, no completed — the server + # Stream ends here on purpose: no task events, no completed —?the server # disconnects after the negotiation Message. The task is started later by # a separate request, only after agreement. @@ -137,6 +150,13 @@ async def _run_agreed_task(self, context: RequestContext, event_queue: EventQueu """Post-agreement task execution: the task is created and runs to completion.""" task_id = context.task_id or "T-negotiation-agreed" context_id = context.context_id or "C-negotiation" + await event_queue.enqueue_event( + Task( + id=task_id, + context_id=context_id, + status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED), + ) + ) for state in (TaskState.TASK_STATE_SUBMITTED, TaskState.TASK_STATE_WORKING): await event_queue.enqueue_event( TaskStatusUpdateEvent(task_id=task_id, context_id=context_id, status=TaskStatus(state=state)) @@ -187,7 +207,7 @@ def build_app(*, scenario: str = "native") -> Starlette: push_store = InMemoryPushNotificationConfigStore() push_http = httpx.AsyncClient(base_url=f"http://127.0.0.1:{port}", trust_env=False) handler = A2ATRequestHandlerDecorator( - LegacyRequestHandler( + DefaultRequestHandler( agent_executor=executor, task_store=InMemoryTaskStore(), agent_card=agent_card, @@ -231,3 +251,4 @@ def main() -> None: if __name__ == "__main__": main() + diff --git a/pyproject.toml b/pyproject.toml index b8478014..b942d115 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,10 +65,6 @@ integration = [ Homepage = "https://github.com/project-openan/a2a-t-sdk-python" Documentation = "https://github.com/project-openan/a2a-t-sdk-python#readme" -[[tool.uv.index]] -url = "https://pypi.tuna.tsinghua.edu.cn/simple" -default = true - [tool.uv.build-backend] module-name = "a2a_t" source-exclude = [".gitignore"] diff --git a/src/a2a_t/observability/server/handler_decorator.py b/src/a2a_t/observability/server/handler_decorator.py index 4351acc6..1b69f929 100644 --- a/src/a2a_t/observability/server/handler_decorator.py +++ b/src/a2a_t/observability/server/handler_decorator.py @@ -534,7 +534,17 @@ async def send_notification(self, task_id: str, event: Any) -> Any: if not self._decorator_active(): return await self._inner.send_notification(task_id, event) info = classify_event(_unwrap(event)) - self._log_push(task_id, info) + dispatch_bearing = _safe_getattr(self._inner, "_dispatch_notification") is not None + if not dispatch_bearing: + # Only the plain-ABC path logs here: dispatch-bearing inners emit one + # log per push_info inside their wrapped _dispatch_notification. + self._log_push(task_id, info) + # V2/core senders route fanout through ``_dispatch_notification`` (whose + # wrapper already emits the per-push_info span with the URL) - forward + # without a second span. Plain ABC senders (no dispatch method) get the + # span here, with the URL read off the event (when present). + if dispatch_bearing: + return await self._inner.send_notification(task_id, event) span = self._start_push_span_for_url(task_id, info, _safe_getattr(event, "url")) cm: Any = _safe_use_span(span) try: @@ -608,25 +618,6 @@ def __init__(self, inner: Any, *, config: A2ATObservabilityConfig | None = None) # plumbing # ------------------------------------------------------------------ - def _warn_if_v2_handler(self) -> None: - """One-time WARNING for a2a-sdk 1.1.x V2 ``DefaultRequestHandler`` inners. - - The V2 implementation's ``ActiveTaskRegistry`` captures the original - executor/push-sender references at construction, so post-construction - wraps (``agent_executor`` / ``_push_sender`` attribute swaps) are silently - bypassed — per-event spans and push LINKs are lost while the entry spans - keep working. ``LegacyRequestHandler`` keeps full observability. - """ - if getattr(self, "_v2_warned", False): - return - if _safe_getattr(self._inner, "_active_task_registry") is None: - return - self._v2_warned = True - logger.warning( - "a2at: DefaultRequestHandler V2 bypasses post-construction executor/push wraps; " - "use LegacyRequestHandler for full per-event/push observability" - ) - def _decorator_active(self) -> bool: """Master gate (spec §8.1): full pass-through only when OTel is unavailable/off. @@ -647,14 +638,29 @@ def __getattr__(self, name: str) -> Any: return getattr(inner, name) def _wrap_collaborators(self) -> None: - """Wrap the inner handler's ``agent_executor``/``_push_sender`` attrs (DefaultRequestHandler shape).""" + """Wrap the inner handler's collaborators for both handler shapes. + + - ``LegacyRequestHandler``: per-attribute ``setattr`` is sufficient (the + handler resolves ``agent_executor`` / ``_push_sender`` per access). + - ``DefaultRequestHandlerV2`` (``_active_task_registry`` present): the + registry captured RAW references at construction, so setattr alone is + bypassed - rebuild the registry with the DECORATED executor / + push sender (D9). The registry is request-state-free at decoration + time, so the rebuild loses nothing. + """ executor = _safe_getattr(self._inner, "agent_executor") + push_sender = _safe_getattr(self._inner, "_push_sender") + if _safe_getattr(self._inner, "_active_task_registry") is None: + self._wrap_legacy_collaborators(executor, push_sender) + return + self._wrap_v2_collaborators(executor, push_sender) + + def _wrap_legacy_collaborators(self, executor: Any, push_sender: Any) -> None: if executor is not None and not isinstance(executor, A2ATAgentExecutorDecorator): try: self._inner.agent_executor = A2ATAgentExecutorDecorator(executor, config=self._config) except Exception: # noqa: BLE001 logger.warning("a2at: failed to wrap agent_executor", exc_info=True) - push_sender = _safe_getattr(self._inner, "_push_sender") if push_sender is not None and not isinstance(push_sender, A2ATPushSenderDecorator): try: decorator = A2ATPushSenderDecorator(push_sender, config=self._config) @@ -662,7 +668,34 @@ def _wrap_collaborators(self) -> None: self._inner._push_sender = decorator except Exception: # noqa: BLE001 logger.warning("a2at: failed to wrap push sender", exc_info=True) - self._warn_if_v2_handler() + + def _wrap_v2_collaborators(self, executor: Any, push_sender: Any) -> None: + """D9: rebuild the V2 ``ActiveTaskRegistry`` with decorated collaborators.""" + try: + if executor is not None and not isinstance(executor, A2ATAgentExecutorDecorator): + executor = A2ATAgentExecutorDecorator(executor, config=self._config) + self._inner.agent_executor = executor + if push_sender is not None and not isinstance(push_sender, A2ATPushSenderDecorator): + push_sender = A2ATPushSenderDecorator(push_sender, config=self._config) + self._inner._push_sender = push_sender + if not push_sender.install_dispatch_wrapper(): + logger.debug( + "a2at: V2 push sender has no _dispatch_notification; " + "push spans will be emitted via the send_notification surface" + ) + registry = _safe_getattr(self._inner, "_active_task_registry") + registry_cls = type(registry) + self._inner._active_task_registry = registry_cls( + agent_executor=executor, + task_store=_safe_getattr(self._inner, "task_store"), + push_sender=push_sender, + ) + logger.info("a2at: DefaultRequestHandler V2 detected - registry rebuilt with decorated collaborators") + except Exception: # noqa: BLE001 + logger.warning( + "a2at: V2 registry rebuild failed; entry spans keep working, per-event/push spans are degraded", + exc_info=True, + ) def _start_server_entry_span(self, name: str, params: Any, context: Any) -> Any | None: if not _otel_compat.is_trace_enabled() or not self._config.trace_enabled: diff --git a/tests/integration/test_e2e_observability.py b/tests/integration/test_e2e_observability.py index bee05ef0..4a07a9c8 100644 --- a/tests/integration/test_e2e_observability.py +++ b/tests/integration/test_e2e_observability.py @@ -1,9 +1,8 @@ """End-to-end observability verification for the v3.0 decorator architecture (spec §10.2, assertions 1-10). In-process wiring: httpx ASGITransport -> Starlette (a2a-python REST routes) -> -A2ATRequestHandlerDecorator(LegacyRequestHandler(...)) on the server and a real -a2a ClientFactory client with register_client_factory(factory) on the client — -zero business observability code. No ASGI middleware: the server entry span +A2ATRequestHandlerDecorator(DefaultRequestHandler(...)) on the server and a real +a2a ClientFactory client with register_client_factory(factory) on the client —?zero business observability code. No ASGI middleware: the server entry span extracts the traceparent from ``ServerCallContext.state["headers"]`` (spec appendix #5), which the client transport decorator injected into ``context.service_parameters``. @@ -12,17 +11,15 @@ Verified against a2a-sdk 1.1.2 structural reality (differs from the unit-test stubs): -- LegacyRequestHandler instead of DefaultRequestHandlerV2: the V2 handler's - ActiveTaskRegistry captures the raw agent_executor/push_sender references at - construction, so post-construction structural wrapping (the executor queue - wrap and the push ``_dispatch_notification`` wrap) would be bypassed; - LegacyRequestHandler resolves ``self.agent_executor`` / ``self._push_sender`` - per call, which is what the decorator replaces. +- The official default ``DefaultRequestHandler`` (a2a-sdk 1.1.x alias for + ``DefaultRequestHandlerV2``) is covered here: the decorator detects the V2 + ``_active_task_registry`` and rebuilds it with the decorated executor / + push sender, so per-event and push spans work on the official handler. + (``LegacyRequestHandler`` remains supported for older a2a-sdk lineages.) - Streaming transport/handler methods are async generators: a2a's trace_class applies its sync wrapper, so a2a's own protocol spans end at generator creation. Because ASGITransport runs the server app inside the client entry - span activation, ambient propagation coincides with the traceparent path — - both route through the same client entry span. + span activation, ambient propagation coincides with the traceparent path —? both route through the same client entry span. - TaskStatusUpdateEvent has no ``final`` field; the terminal state itself is final. @@ -96,10 +93,10 @@ def _agent_card(push: bool) -> Any: def _build_app(executor_kind: str, *, config: Any = None, push: bool = False) -> tuple[Any, Any]: - """Starlette app: A2ATRequestHandlerDecorator(LegacyRequestHandler(...)) + a2a REST routes.""" + """Starlette app: A2ATRequestHandlerDecorator(DefaultRequestHandler(...)) + a2a REST routes.""" import httpx from a2a.server.agent_execution import AgentExecutor - from a2a.server.request_handlers.default_request_handler import LegacyRequestHandler + from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.routes import create_agent_card_routes, create_rest_routes from a2a.server.tasks.base_push_notification_sender import BasePushNotificationSender from a2a.server.tasks.inmemory_push_notification_config_store import InMemoryPushNotificationConfigStore @@ -108,6 +105,7 @@ def _build_app(executor_kind: str, *, config: Any = None, push: bool = False) -> Artifact, Message, Role, + Task, TaskArtifactUpdateEvent, TaskState, TaskStatus, @@ -121,6 +119,14 @@ class FlowExecutor(AgentExecutor): async def execute(self, context: Any, event_queue: Any) -> None: task_id = context.task_id context_id = context.context_id + # V2 protocol: the executor enqueues the initial Task snapshot first + await event_queue.enqueue_event( + Task( + id=task_id, + context_id=context_id, + status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED), + ) + ) for state in (TaskState.TASK_STATE_SUBMITTED, TaskState.TASK_STATE_WORKING): await event_queue.enqueue_event( TaskStatusUpdateEvent(task_id=task_id, context_id=context_id, status=TaskStatus(state=state)) @@ -185,7 +191,7 @@ async def cancel(self, context: Any, event_queue: Any) -> None: ... handler_kwargs["push_config_store"] = push_store handler_kwargs["push_sender"] = BasePushNotificationSender(push_http, push_store) - handler = LegacyRequestHandler( + handler = DefaultRequestHandler( agent_executor=executors[executor_kind], task_store=InMemoryTaskStore(), agent_card=agent_card, @@ -377,7 +383,7 @@ async def test_e2e_trace_continuity_event_spans_and_metrics(otel_setup: tuple[An events = await _send_streaming(client, {_TASK_T_URI: "prompt-text"}, context_id="C-e2e") finally: await httpx_client.aclose() - assert len(events) == 4 + assert len(events) == 5 # V2: + initial Task snapshot spans = _a2at_spans(exporter.get_finished_spans()) client_entry = _find(spans, "SendStreamingMessage", side="client") @@ -419,8 +425,8 @@ async def test_e2e_trace_continuity_event_spans_and_metrics(otel_setup: tuple[An for span in spans if span.name == "SendStreamingMessage-event" and span.kind is SpanKind.SERVER ] - assert len(client_events) == 4 - assert len(server_events) == 4 + assert len(client_events) == 5 + assert len(server_events) == 5 assert all(span.kind is SpanKind.CLIENT for span in client_events) assert all( span.links and link.context.span_id == client_entry.context.span_id for span in client_events for link in span.links @@ -436,6 +442,7 @@ async def test_e2e_trace_continuity_event_spans_and_metrics(otel_setup: tuple[An "status", "artifact", "completed", + "task", } # (8) metrics: L1 gen_ai histogram (client) + L3 a2at histogram with client AND server side attribution @@ -566,7 +573,7 @@ async def test_e2e_push_notification_span(otel_setup: tuple[Any, Any]) -> None: ) finally: await httpx_client.aclose() - assert len(events) == 4 + assert len(events) == 5 # V2: + initial Task snapshot spans = _a2at_spans(exporter.get_finished_spans()) server_entry = _find(spans, "SendStreamingMessage", side="server") @@ -596,7 +603,7 @@ async def test_e2e_payload_attributes_and_log_trace_context( events = await _send_streaming(client, {_TASK_T_URI: "prompt-text"}, context_id="C-log") finally: await httpx_client.aclose() - assert len(events) == 4 + assert len(events) == 5 # V2: + initial Task snapshot spans = _a2at_spans(exporter.get_finished_spans()) client_entry = _find(spans, "SendStreamingMessage", side="client") @@ -637,7 +644,7 @@ async def test_e2e_master_switch_off( events = await _send_streaming(client, {_TASK_T_URI: "prompt"}, context_id="C-off") finally: await httpx_client.aclose() - assert len(events) == 4 + assert len(events) == 5 # V2: + initial Task snapshot spans = exporter.get_finished_spans() assert spans @@ -684,3 +691,50 @@ async def test_register_client_factory_decorates_real_factory_transport() -> Non assert isinstance(client2._transport._inner, RestTransport) finally: await httpx_client.aclose() + + + + + + + +async def test_sample_app_smoke_on_official_handler(otel_setup: tuple[Any, Any]) -> None: + """Smoke guard: the observability-sample app must work on the official + DefaultRequestHandler — the V2 protocol requires the executor to enqueue + the initial Task snapshot before any status event (regression guard for + the sample executors, which have no automated coverage otherwise).""" + import sys + from pathlib import Path as _Path + + sample_src = _Path(__file__).resolve().parents[2] / "a2a-t-sample" / "observability-sample" / "src" + sys.path.insert(0, str(sample_src)) + try: + from a2a.types import AgentCard + from google.protobuf.json_format import ParseDict + from obs_sample.server_main import build_agent_card_payload, build_app + finally: + sys.path.remove(str(sample_src)) + + exporter, _reader = otel_setup + app = build_app(scenario="negotiation") + card = ParseDict(build_agent_card_payload(8100), AgentCard()) + client, httpx_client = _make_client(app, card) + try: + # The sample server matches the Negotiation-T URI WITH the /v1 suffix + events = await _send_streaming(client, {_NEG_T_URI + "/v1": "offer"}, context_id="C-smoke") + finally: + await httpx_client.aclose() + assert events and events[-1].HasField("message") + await _finalize_abandoned_generators() + + spans = _a2at_spans(exporter.get_finished_spans()) + client_negotiations = [ + s for s in spans if s.name == "SendStreamingMessage-negotiation" and s.kind is SpanKind.CLIENT + ] + server_negotiations = [ + s for s in spans if s.name == "SendStreamingMessage-negotiation" and s.kind is SpanKind.SERVER + ] + assert len(client_negotiations) == 1 + assert len(server_negotiations) == 1 + assert client_negotiations[0].parent is not None + assert server_negotiations[0].parent is not None diff --git a/tests/observability/test_handler_decorator.py b/tests/observability/test_handler_decorator.py index 1bd6efdc..fc2c41ad 100644 --- a/tests/observability/test_handler_decorator.py +++ b/tests/observability/test_handler_decorator.py @@ -807,29 +807,66 @@ async def test_subscribe_to_task_stream_entry_span(exporter: InMemorySpanExporte assert spans[0].status.status_code == StatusCode.OK -async def test_v2_handler_emits_one_time_degradation_warning(caplog: pytest.LogCaptureFixture) -> None: - """a2a-sdk 1.1.2+ V2 DefaultRequestHandler (``_active_task_registry``): post-construction - executor/push wraps are silently bypassed — the decorator warns once per instance.""" +class _FakeV2Registry: + """Accepts the ActiveTaskRegistry constructor kwargs, records them.""" + + def __init__(self, **kwargs: Any) -> None: + self.created_with = kwargs + + +class _StrictRegistry: + """Constructor signature incompatible with the rebuilt registry (degradation path).""" + + def __init__(self) -> None: ... + + +async def test_v2_handler_registry_rebuilt_with_decorated_collaborators( + exporter: InMemorySpanExporter, caplog: pytest.LogCaptureFixture +) -> None: + """D9：V2 handler（``_active_task_registry`` 存在）→ 装饰器用装饰版 executor/push_sender + 重建 registry（构造期裸引用被替换），并打 INFO 适配日志。""" + import logging + inner = FakeSimpleHandler(result=FakeMessage(metadata={}, task_id="T26", context_id="C26")) - inner._active_task_registry = object() + inner.agent_executor = FakeExecutor(events=[]) + inner._push_sender = FakeDispatchPushSender(["https://hook.example"]) + inner._active_task_registry = _FakeV2Registry() - with caplog.at_level(logging.WARNING, logger="a2at.observability"): - decorator = A2ATRequestHandlerDecorator(inner) - decorator._wrap_collaborators() # re-wrap within the same instance: warning stays one-time + with caplog.at_level(logging.INFO, logger="a2at.observability"): + A2ATRequestHandlerDecorator(inner) - warnings = [record for record in caplog.records if "LegacyRequestHandler" in record.getMessage()] - assert len(warnings) == 1 - assert warnings[0].levelno == logging.WARNING + assert isinstance(inner.agent_executor, A2ATAgentExecutorDecorator) + assert isinstance(inner._push_sender, A2ATPushSenderDecorator) + registry = inner._active_task_registry + assert isinstance(registry, _FakeV2Registry) + assert registry.created_with["agent_executor"] is inner.agent_executor + assert registry.created_with["push_sender"] is inner._push_sender + assert any("V2" in record.getMessage() and "rebuilt" in record.getMessage() for record in caplog.records) -async def test_legacy_handler_does_not_warn(caplog: pytest.LogCaptureFixture) -> None: - """LegacyRequestHandler-shaped inners (no ``_active_task_registry``) stay silent.""" - inner = FakeSimpleHandler(result=FakeMessage(metadata={}, task_id="T27", context_id="C27")) +async def test_v2_registry_rebuild_failure_degrades_gracefully( + exporter: InMemorySpanExporter, caplog: pytest.LogCaptureFixture +) -> None: + """registry 签名不兼容（版本变化）→ WARNING 降级：入口 span 仍工作（spec §8.2）。""" + import logging + + inner = FakeStreamingHandler(FakeExecutor(events=[])) + inner._push_sender = FakeDispatchPushSender(["https://hook.example"]) + inner._active_task_registry = _StrictRegistry() with caplog.at_level(logging.WARNING, logger="a2at.observability"): A2ATRequestHandlerDecorator(inner) - assert not [record for record in caplog.records if "LegacyRequestHandler" in record.getMessage()] + assert any("V2 registry rebuild failed" in record.getMessage() for record in caplog.records), ( + "expected the rebuild-failure WARNING" + ) + + # D9 test-plan item 5: 入口 span 仍工作——一次流式请求仍产生 SERVER 入口 span + decorator = A2ATRequestHandlerDecorator(inner) + params = make_send_request({}, task_id="T26c", context_id="C26c") + _ = [event async for event in decorator.on_message_send_stream(params, FakeServerCallContext())] + entry = [s for s in exporter.get_finished_spans() if s.name == "SendStreamingMessage"] + assert len(entry) == 1 and entry[0].kind is SpanKind.SERVER async def test_stream_negotiation_message_creates_server_negotiation_span(