From 846397d41f5c89c5527756e345bbeaebb41d5423 Mon Sep 17 00:00:00 2001 From: zhangwenjun-huawei Date: Mon, 26 Oct 2026 01:57:04 +0800 Subject: [PATCH] docs: add the observability integration guide (en+zh) - developer guide: new 1.7 chapter covering the capability overview, the client/server integration steps, the auto-recorded traces, metrics and logs, the environment variable reference, and the optional manual entry points; the configuration reference moves from 1.7 to 1.8 - READMEs: one capability row and one project-structure row for the observability module Signed-off-by: zhangwenjun-huawei --- README.md | 4 +- README_zh.md | 4 +- docs/en/developer_guide.md | 208 +++++++++++++++++- ...00\345\217\221\346\214\207\345\215\227.md" | 208 +++++++++++++++++- 4 files changed, 420 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index b50bfe37..42a58687 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,7 @@ flowchart LR | Negotiation content API | Supports `information` / `feasibility` / `target` negotiation types plus `abort` termination messages, with template-driven negotiation message generation and validation; negotiation session state travels in the message metadata (`negotiationContext`) and the SDK itself is stateless | | Resource organization | Bundled prompt resources (`prompts` / `scenarios` / `slots` / `templates` / `negotiation-vocabulary`) ship with the package, supporting both `packaged` built-in resources and `local_file` local files | | LLM adaptation | Connects to external LLMs through OpenAI-compatible call chains, retrying retryable error codes up to the configured attempt limit | +| Observability | Automatically records traces, duration metrics, and structured logs of A2A-T interactions based on OpenTelemetry; integrate with one registration on the client and one wrapper on the server, with no business code changes, and it can be switched off entirely as needed | | Bundled samples | The repository ships runnable sample scenarios such as `subscribe_incident` (event subscription); without an API key they automatically degrade to scripted mock LLM responses, running end to end with zero external dependencies | ## Project Structure @@ -82,6 +83,7 @@ The repository is organized with `uv`; the core code lives under `src/a2a_t`: | `llm` | LLM adaptation layer with a default OpenAI-compatible client; supports custom LLM integration | | `prompt` | Prompt resource analysis, slot extraction, template rendering, and validation | | `negotiation` | Negotiation content models, the generation pipeline, and the validation pipeline | +| `observability` | The observability module, automatically recording interaction traces, metrics, and logs (requires the `a2a-t-sdk[observability]` extra dependency) | | `a2a-t-sample` | Runnable client/server sample case collection | | `a2a-t-corpus` | Accuracy verification corpus (pure test assets): data-driven workflow cases | @@ -159,7 +161,7 @@ More runnable samples (event subscription end-to-end, RAN energy saving with neg | Document | Location | Content | | --- | --- | --- | -| Developer Guide | [docs/en/developer_guide.md](docs/en/developer_guide.md) | Feature introduction, installation and integration, parameter configuration, and minimal practices | +| Developer Guide | [docs/en/developer_guide.md](docs/en/developer_guide.md) | Feature introduction, installation and integration, parameter configuration, observability integration, and minimal practices | | API Reference | [docs/en/API_Reference.md](docs/en/API_Reference.md) | Full API definitions and usage of `A2ATClient` / `A2ATServer` | ## Current Support Scope diff --git a/README_zh.md b/README_zh.md index 615cffa9..d128bf0d 100644 --- a/README_zh.md +++ b/README_zh.md @@ -66,6 +66,7 @@ flowchart LR | 协商内容 API | 支持 `information` / `feasibility` / `target` 三类协商及 `abort` 终止消息，提供模板驱动的协商报文生成与校验；协商会话状态随消息 metadata（`negotiationContext`）往返，SDK 本身无状态 | | 资源组织 | 内置提示词资源（`prompts` / `scenarios` / `slots` / `templates` / `negotiation-vocabulary`）随包提供，支持 `packaged` 内置资源与 `local_file` 本地文件两种加载方式 | | LLM 适配 | 通过 OpenAI 兼容调用链接入外部大模型；对可重试错误码按配置的尝试次数上限重试 | +| 可观测性 | 基于 OpenTelemetry 自动记录 A2A-T 交互的调用链、耗时指标与结构化日志；客户端注册一次、服务端包装一次即可接入，无需修改业务代码，支持按需整体关闭 | | 内置示例 | 随仓库提供 `subscribe_incident`（事件订阅）等可运行示例场景，未配置 API Key 时自动降级为脚本化 Mock LLM，零外部依赖即可端到端跑通 | ## 项目结构 @@ -82,6 +83,7 @@ flowchart LR | `llm` | LLM 适配层，默认提供 OpenAI 兼容客户端，支持自定义 LLM 接入 | | `prompt` | 提示词资源分析、槽位提取、模板渲染与校验 | | `negotiation` | 协商内容模型、生成管线与校验管线 | +| `observability` | 可观测性模块，自动记录交互调用链、指标与日志（需安装 `a2a-t-sdk[observability]` 额外依赖） | | `a2a-t-sample` | 可运行的客户端/服务端示例用例集 | | `a2a-t-corpus` | 准确性验证语料（纯测试资产）：数据驱动工作流用例 | @@ -157,7 +159,7 @@ uv run mypy src # 类型检查 | 文档 | 位置 | 内容 | | --- | --- | --- | -| 开发指南 | [docs/zh/开发指南.md](docs/zh/开发指南.md) | 特性介绍、安装接入、参数配置与最小实践 | +| 开发指南 | [docs/zh/开发指南.md](docs/zh/开发指南.md) | 特性介绍、安装接入、参数配置、可观测性集成与最小实践 | | API 参考 | [docs/zh/API参考.md](docs/zh/API参考.md) | `A2ATClient` / `A2ATServer` 全量 API 定义与使用说明 | ## 当前支持范围 diff --git a/docs/en/developer_guide.md b/docs/en/developer_guide.md index f6853a93..0817ddc7 100644 --- a/docs/en/developer_guide.md +++ b/docs/en/developer_guide.md @@ -23,6 +23,7 @@ Main capabilities include: - **Template resource management**: built-in scenario, slot, template, vocabulary, and system prompt resources, supporting both the `packaged` built-in source and the `local_file` local-file source. - **LLM adaptation**: connects to external large language models through OpenAI-compatible call chains. - **Structured, bilingual error model**: 42 fixed machine-readable error codes (with the parameters needed to render their messages) and bilingual message templates. +- **Observability**: automatically records traces, duration metrics, and structured logs of A2A interactions based on OpenTelemetry; one integration step on each of the client and the server, with no business code changes (see 1.7). For the complete API list and usage, see [API_Reference.md](API_Reference.md). @@ -1291,7 +1292,212 @@ For daily rotation, use `TimedRotatingFileHandler("logs/llm-call.log", when="mid > Note: the SDK only produces log events and configures no handler/level; the log output path and rotation policy are entirely decided by the integrator's configuration (the examples above are suggestions only). Once `A2AT_LLM_DETAIL_LOG_ENABLED` is enabled, every LLM call outputs its complete message; set per-file size limits and retention counts as in the examples to avoid filling the disk. If the integrator does not configure logging at all, Python by default sends only WARNING and above to stderr, so the DEBUG-level `llm_call` logs are invisible. -## 1.7 Configuration Reference +## 1.7 Observability Integration Guide + +### 1.7.1 Capability Overview + +Purpose: answer the question "what happened after the request went out" during joint debugging and O&M — who called whom, which step the task has reached, how long each step took, and how many rounds a negotiation went through. The SDK's observability module is based on OpenTelemetry (the industry-standard observability data framework) and automatically records the A2A interaction process. The recorded data falls into three categories: + +| Data type | Description | Typical use | +| --- | --- | --- | +| Traces | The complete course of one interaction from the client initiating the request to the server finishing the processing, broken down into per-step records: the client sending the request, the server handling the request, and each streamed event are each a record | Troubleshoot "where the request is stuck" and confirm whether the records from both ends correlate | +| Metrics | Time-aggregated values, such as the task request duration and the number of negotiation rounds to convergence | Observe overall trends and configure alerts | +| Logs | Structured records (JSON) of key events, automatically carrying the trace identity | Look up the detailed messages and status changes of one interaction by the trace identity | + +Integration: **register once on the client `ClientFactory` and wrap the request handler once on the server**; afterwards the SDK records the data above automatically — no business code changes and no hand-written OpenTelemetry code are needed. Without a data collection endpoint, the records go to the console (directly visible during joint debugging); with an endpoint configured, they are exported to an OpenTelemetry collector and viewed in a backend (such as Jaeger or Langfuse). + +Notes: + +1. Observability is an optional capability installed through the extra dependency `a2a-t-sdk[observability]`; without it the SDK behavior is unaffected. +2. The observability logic has zero impact on the business flow: a recording failure only produces a warning log and never interrupts request processing; it can be switched off entirely or per data type. +3. Difference from section 1.6: section 1.6 covers the SDK-internal LLM call logs (for prompt tuning), while this section covers the full A2A interaction chain, whose data can be exported to a unified observability backend. + +### 1.7.2 Integration Steps + +#### Step1 Install Dependencies + +The observability module's code ships with `a2a-t-sdk` — no separate package is needed; however, the OpenTelemetry components it relies on at runtime are optional dependencies that `pip install a2a-t-sdk` does not install — without them the observability logic silently disables itself (no records are produced and the business flow is unaffected), so the extra dependency is needed to fill the gap: + +```bash +pip install a2a-t-sdk[observability] +``` + +> If the environment already has the OpenTelemetry components installed (for example, other observability capabilities are integrated), this step can be skipped. + +#### Step2 Client Integration + +Before creating clients with the official `ClientFactory`, register the factory once; every client produced by that factory is then observed automatically: + +```python +from a2a.client.client import ClientConfig +from a2a.client.client_factory import ClientFactory +from a2a.utils.constants import TransportProtocol +from a2a_t.observability import register_client_factory + +factory = ClientFactory( + ClientConfig( + supported_protocol_bindings=[TransportProtocol.HTTP_JSON], + use_client_preference=True, + ) +) +register_client_factory(factory) # register once; no other business code changes +a2a_client = factory.create(agent_card) +``` + +After registration the client automatically: records the request initiation, records each streamed event it receives, and injects the trace identity (`traceparent`) into the request headers to correlate with the server-side records. + +#### Step3 Server Integration + +Wrap the official request handler with `A2ATRequestHandlerDecorator`, then assemble the routes as usual: + +```python +from a2a.server.request_handlers import DefaultRequestHandler +from a2a.server.routes import create_agent_card_routes, create_rest_routes +from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore +from a2a_t.observability import A2ATRequestHandlerDecorator +from starlette.applications import Starlette + +request_handler = A2ATRequestHandlerDecorator( # one wrapper around the original handler + DefaultRequestHandler( + agent_executor=executor, + task_store=InMemoryTaskStore(), + agent_card=agent_card, + ) +) + +app = Starlette(routes=[ + *create_agent_card_routes(agent_card), + *create_rest_routes(request_handler), +]) +``` + +After wrapping, the server automatically: records the request handling (extracting the trace identity from the request headers to join the client records into the same trace), records each streamed event it sends, and records push notification deliveries. + +> Both the official `DefaultRequestHandler` (a2a-sdk 1.1.x) and the `LegacyRequestHandler` can be wrapped directly. + +#### Step4 Configure the Data Destination + +Observability configuration is read from **OS environment variables** (note the difference from section 1.8, where the business configuration is read from the `.env` file). The SDK completes its initialization automatically on the first observed call. Common settings: + +| Environment variable | Default | Description | +| --- | --- | --- | +| `A2AT_OBSERVABILITY_ENABLED` | `true` | Master switch; set to `false` and the module becomes fully transparent, producing no records | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | unset | Data collection endpoint address. When unset, records are printed to the console for direct inspection during joint debugging | +| `OTEL_SERVICE_NAME` (or `A2AT_SERVICE_NAME`) | `a2a-t-agent` | The service name of this process shown in the backend; use different values for the client and the server | +| `A2AT_EXPORTER_PROTOCOL` | `grpc` | Export protocol; `http/protobuf` is also supported | + +Notes: + +1. If the application has already initialized OpenTelemetry itself (for example, other observability components are integrated), the SDK does not override the existing configuration. +2. Initialization can also be done explicitly in code: `from a2a_t.observability import setup; setup(endpoint=..., service_name=...)`; the passed arguments take precedence over environment variables. +3. For the complete configuration list (per-data-type switches, message payload logging switches, etc.), see 1.7.4. + +#### Step5 Verify the Integration + +Without `OTEL_EXPORTER_OTLP_ENDPOINT` configured, both consoles print the JSON of every trace record for direct inspection. The repository ships a runnable end-to-end sample (with three scenarios — task dispatch, LLM prompt generation, and multi-round negotiation — plus instructions for connecting a Langfuse backend); run the sample before integrating into your own project: + +```bash +cd a2a-t-sample +# PowerShell: $env:PYTHONPATH = "$PWD\observability-sample\src" +export PYTHONPATH="$PWD/observability-sample/src" +uv run python -m obs_sample.server_main # terminal 1: the server (port 8100) +uv run python -m obs_sample.client_main --scenario=native # terminal 2: the client +``` + +For details, see [observability-sample/README.md](../../a2a-t-sample/observability-sample/README.md). + +### 1.7.3 What Gets Recorded Automatically + +After integration, no observability code is needed; the SDK records automatically per interaction action: + +| Interaction action | What is recorded automatically | +| --- | --- | +| The client initiates a request (SendMessage / SendStreamingMessage, etc.) | The client entry record, with the trace identity injected into the request | +| The server receives and handles the request | The server entry record, joined with the client entry record into the same trace | +| The server streams out events (status changes, result artifacts, completion, etc.) | One record per event, carrying the event kind (`streaming.event.kind`) and the task status (`task.status`) | +| Negotiation messages (Negotiation-T) sent or received | A negotiation record carrying the negotiation identity (`negotiation.id`), the current round (`negotiation.round`), and the negotiation action (`negotiation.performative`); when the negotiation reaches agreement or terminates, the converged-rounds metric `a2at.negotiation.total_rounds` is reported automatically | +| The server delivers a push notification | A push record (`SendMessage-pushNotification`) | +| A task request finishes processing | The duration metric `a2at.task.request.duration` (recorded on both ends, told apart by the `a2at.span.side` attribute) | +| The LLM is called through an SDK facade such as `A2ATClient` | The input/output token usage is recorded automatically onto the subsequent trace | +| Key events (task status changes, result artifacts, etc.) | Structured JSON logs on the logger `a2at.observability`, automatically carrying `trace_id` / `span_id` for looking up logs by the trace identity | + +The trace records also automatically carry A2A-T business attributes (prefixed with `gen_ai.agent.a2at.`), which the backend can filter and aggregate by. Common attributes: + +| Attribute | Description | Source | +| --- | --- | --- | +| `extension.name` | The A2A-T extension type of this interaction (Task-T / Negotiation-T / Notification-T / Authorization-T) | Extracted from the extension URI in the message metadata | +| `task.type` | The task type (for example "configuration dispatch") | Extracted from the message body by the template section heading | +| `notification.topic` | The notification subscription topic | Same as above | +| `authorization.policy.operation.type` | The operation type of the authorization policy | Same as above | +| `negotiation.id` / `negotiation.round` | The negotiation identity and the current round | Extracted from the `negotiationContext` in the message metadata | + +For example, filtering by `extension.name=Task-T` in the backend shows only the traces of task-dispatch interactions; querying by `negotiation.id` reconstructs the complete course of a multi-round negotiation. + +### 1.7.4 Configuration Reference + +All observability configuration items are read from OS environment variables (not from the `.env` file): + +| Environment variable | Default | Description | +| --- | --- | --- | +| `A2AT_OBSERVABILITY_ENABLED` | `true` | Master switch; `false` makes the module fully transparent | +| `A2AT_TRACE_ENABLED` / `A2AT_METRIC_ENABLED` / `A2AT_LOG_ENABLED` | `true` | Per-data-type switches; for example, metrics and logs keep working with traces switched off | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | unset | Data collection endpoint; when unset, records go to the console (joint-debugging mode) | +| `A2AT_CONSOLE_EXPORTER` | `true` | Set to `false` to turn off the console output of the joint-debugging mode (no output at all when no collection endpoint is configured) | +| `OTEL_SERVICE_NAME` (or `A2AT_SERVICE_NAME`) | `a2a-t-agent` | Service name | +| `A2AT_AGENT_ROLE` | unset | Process role marker (such as `client` / `server`), written into the resource attribute `agent.role` | +| `A2AT_EXPORTER_PROTOCOL` | `grpc` | Export protocol; `http/protobuf` is also supported | +| `A2AT_EXTRACT_REQUEST` / `A2AT_EXTRACT_RESPONSE` | `false` | Whether to write the request/response message payload into the attributes of the trace records (JSON, truncated per the same rule as the payload logs) | +| `A2AT_LOGS_PAYLOAD_ENABLED` | `false` | Whether to include the message payload in the structured logs | +| `A2AT_LOGS_PAYLOAD_MAX_LENGTH` | `4096` | Maximum length (characters) of one payload entry; the excess is truncated and marked with `[truncated]` | +| `OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED` | `true` | Hard switch; `false` disables all observability logic at module load time | + +> Risk warning: with `A2AT_EXTRACT_REQUEST` / `A2AT_EXTRACT_RESPONSE` or `A2AT_LOGS_PAYLOAD_ENABLED` enabled, the request and response payloads (including the prompt content) enter the trace attributes or the logs and may contain sensitive information — use them only during joint debugging and keep them off in production; long payloads also increase the data volume significantly, so control them with `A2AT_LOGS_PAYLOAD_MAX_LENGTH`. For redaction, pass a custom `payload_redactor` function when constructing `A2ATObservabilityConfig` (redaction runs before truncation), and pass the config through `register_client_factory(factory, config=...)` or `A2ATRequestHandlerDecorator(handler, config=...)`. + +### 1.7.5 Adding Observability Data in Business Code (Optional) + +Beyond the automatic records, the SDK provides three additional entry points for use as needed: + +**1. Add business attributes to the current step** + +In business code such as `AgentExecutor.execute`, call `current_span()` to get a view of the current step and append custom attributes (for example, a business order number): + +```python +from a2a_t.observability import current_span + +span = current_span() +if span is not None: + span.set_attribute("biz.order_id", order_id) +``` + +It returns `None` when observability is not integrated or no trace is in progress; the caller just checks for `None`. + +**2. Record a piece of business processing manually** + +Use `A2ATSpan` to record the duration and attributes of a piece of business processing: + +```python +from a2a_t.observability import A2ATSpan + +with A2ATSpan("query-ne-config") as span: + span.set_attribute("ne.name", ne_name) + query_ne_config(ne_name) +``` + +**3. Record SDK facade calls** + +Wrap an `A2ATClient` / `A2ATServer` instance with `trace_facade` to record one step per SDK-internal processing such as prompt generation and message validation (named like `a2at.sdk.client.generate_task_prompt_from_text`): + +```python +from pathlib import Path +from a2a_t.client.a2at_client import A2ATClient +from a2a_t.observability import trace_facade + +client = trace_facade(A2ATClient(env_path=Path("package_data/.env"))) +``` + +> The entry points above automatically become no-ops (producing no records) when the observability dependencies are not installed or the master switch is off, so the business code needs no extra checks. + +## 1.8 Configuration Reference The sample configuration file provided by the SDK is `env.example`; copy it to `package_data/.env` (or pass an explicit `env_path`). Both the A2A-T Client and the A2A-T Server read the `.env` file at construction (OS environment variables are not read). The configuration items are: diff --git "a/docs/zh/\345\274\200\345\217\221\346\214\207\345\215\227.md" "b/docs/zh/\345\274\200\345\217\221\346\214\207\345\215\227.md" index ee6bf599..4f0fae37 100644 --- "a/docs/zh/\345\274\200\345\217\221\346\214\207\345\215\227.md" +++ "b/docs/zh/\345\274\200\345\217\221\346\214\207\345\215\227.md" @@ -23,6 +23,7 @@ a2a-t-sdk-python是面向电信智能体协作场景的 Python SDK，用于在 A - **模板资源管理**：内置场景、槽位、模板、词表和系统提示词资源，支持`packaged`内置资源与`local_file`本地文件两种资源加载方式。 - **LLM 适配**：通过 OpenAI-compatible 调用链接入外部大模型。 - **结构化双语错误模型**：42 个固定的机器可读错误码（附渲染错误信息所需的参数）及中英文消息模板。 +- **可观测性**：基于 OpenTelemetry 自动记录 A2A 交互的调用链、耗时指标与结构化日志，客户端与服务端各一步接入，无需修改业务代码（详见 1.7）。 SDK提供的API列表和使用说明，详见 [API参考.md](API参考.md)。 @@ -1282,7 +1283,212 @@ call_logger.propagate = False > 说明：SDK 只产生日志事件，不配置任何 handler/level；日志打印路径与绕接策略全部由集成方配置决定（上文示例仅作建议）。开启 `A2AT_LLM_DETAIL_LOG_ENABLED` 后每次 LLM 调用会输出完整报文，建议按示例设置单文件大小上限与保留份数，避免占满磁盘。若集成方完全未配置 logging，Python 默认只把 WARNING 及以上输出到 stderr，DEBUG 级 `llm_call` 日志不可见。 -## 1.7 配置项列表 +## 1.7 可观测性集成指引 + +### 1.7.1 能力概述 + +目的：在联调和运维阶段回答“请求发出去之后发生了什么”——谁调用了谁、任务进行到哪一步、各环节耗时多久、协商进行了几轮。SDK 的可观测性模块基于 OpenTelemetry（业界通用的可观测数据标准）自动记录 A2A 交互过程，记录的数据分三类： + +| 数据类型 | 说明 | 典型用途 | +| --- | --- | --- | +| 调用链 | 一次交互从客户端发起到服务端处理完成的完整经过，按环节拆成若干条记录：客户端发起请求、服务端处理请求、流式返回的每个事件各为一条 | 排查“请求卡在哪一步”、确认两端记录能否关联 | +| 指标 | 按时间聚合的数值，如任务请求耗时、协商收敛轮数 | 观察整体趋势、配置告警 | +| 日志 | 关键事件的结构化记录（JSON 格式），自动携带调用链标识 | 按调用链标识反查一次交互的详细报文与状态变化 | + +接入方式：**客户端对 `ClientFactory` 注册一次、服务端对请求处理器包装一次**，之后 SDK 自动记录上述数据，不需要改动业务代码，也不需要手写 OpenTelemetry 相关代码。未配置数据收集端点时，记录输出到控制台（联调阶段直接可见）；配置端点后上报到 OpenTelemetry 收集器，在后端（如 Jaeger、Langfuse）统一查看。 + +说明： + +1. 可观测性为可选能力，通过额外依赖 `a2a-t-sdk[observability]` 安装；不安装时 SDK 行为不受影响。 +2. 观测逻辑对业务流程零影响：记录失败仅输出告警日志，不会中断请求处理；支持整体关闭或按数据类型关闭。 +3. 本节与 1.6 节的区别：1.6 节是 SDK 内部 LLM 调用日志（面向提示词调优），本节面向 A2A 交互全链路，数据可上报到统一可观测后端。 + +### 1.7.2 接入步骤 + +#### Step1 安装依赖 + +可观测性模块的代码随 `a2a-t-sdk` 一起安装，无需单独安装其他包；但其运行依赖的 OpenTelemetry 组件属于可选依赖，`pip install a2a-t-sdk` 默认不会安装——缺少这些组件时观测逻辑自动失效（不产生任何记录，也不影响业务流程），因此需要通过额外依赖补齐： + +```bash +pip install a2a-t-sdk[observability] +``` + +> 若环境中已安装 OpenTelemetry 相关组件（例如集成了其他可观测能力），可跳过本步骤。 + +#### Step2 客户端接入 + +使用官方 `ClientFactory` 创建客户端之前，对工厂注册一次；之后该工厂创建的所有客户端自动被观测： + +```python +from a2a.client.client import ClientConfig +from a2a.client.client_factory import ClientFactory +from a2a.utils.constants import TransportProtocol +from a2a_t.observability import register_client_factory + +factory = ClientFactory( + ClientConfig( + supported_protocol_bindings=[TransportProtocol.HTTP_JSON], + use_client_preference=True, + ) +) +register_client_factory(factory) # 注册一次即可，业务代码无其他改动 +a2a_client = factory.create(agent_card) +``` + +注册后客户端自动完成：发起请求的调用链记录、流式接收事件的逐条记录、向请求头注入调用链标识（`traceparent`，用于与服务端记录串联）。 + +#### Step3 服务端接入 + +用 `A2ATRequestHandlerDecorator` 包装官方请求处理器，再按原有方式组装路由： + +```python +from a2a.server.request_handlers import DefaultRequestHandler +from a2a.server.routes import create_agent_card_routes, create_rest_routes +from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore +from a2a_t.observability import A2ATRequestHandlerDecorator +from starlette.applications import Starlette + +request_handler = A2ATRequestHandlerDecorator( # 在原处理器外面套一层即可 + DefaultRequestHandler( + agent_executor=executor, + task_store=InMemoryTaskStore(), + agent_card=agent_card, + ) +) + +app = Starlette(routes=[ + *create_agent_card_routes(agent_card), + *create_rest_routes(request_handler), +]) +``` + +包装后服务端自动完成：处理请求的调用链记录（从请求头提取调用链标识，与客户端记录串联为同一条调用链）、流式发出事件的逐条记录、推送通知（Push Notification）的记录。 + +> 官方 `DefaultRequestHandler`（a2a-sdk 1.1.x）与 `LegacyRequestHandler` 均可直接包装。 + +#### Step4 配置数据去向 + +可观测性配置从**操作系统环境变量**读取（注意与 1.8 节业务配置读取 `.env` 文件的方式不同）。首次发生被观测的调用时 SDK 自动完成初始化，常用配置如下： + +| 环境变量 | 默认值 | 说明 | +| --- | --- | --- | +| `A2AT_OBSERVABILITY_ENABLED` | `true` | 总开关；置 `false` 后模块完全透明，不产生任何记录 | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | 未设置 | 数据收集端点地址。未设置时记录输出到控制台，便于联调阶段直接查看 | +| `OTEL_SERVICE_NAME`（或 `A2AT_SERVICE_NAME`） | `a2a-t-agent` | 本进程在后端展示的服务名，建议客户端与服务端设置不同取值 | +| `A2AT_EXPORTER_PROTOCOL` | `grpc` | 上报协议，可选 `http/protobuf` | + +说明： + +1. 应用若已自行初始化 OpenTelemetry（例如集成了其他可观测组件），SDK 不会覆盖已有配置。 +2. 也可在代码中显式初始化：`from a2a_t.observability import setup; setup(endpoint=..., service_name=...)`，传入参数优先于环境变量。 +3. 完整配置项（按数据类型独立开关、报文日志开关等）见 1.7.4。 + +#### Step5 验证接入效果 + +未配置 `OTEL_EXPORTER_OTLP_ENDPOINT` 时，两端控制台会输出每条调用链记录的 JSON，可直接确认记录内容。仓库提供可运行的端到端样例（含任务下发、LLM 生成提示词、多轮协商三个场景，以及 Langfuse 后端的接入说明），建议先跑通样例再接入自有工程： + +```bash +cd a2a-t-sample +# PowerShell：$env:PYTHONPATH = "$PWD\observability-sample\src" +export PYTHONPATH="$PWD/observability-sample/src" +uv run python -m obs_sample.server_main # 终端一：服务端（端口 8100） +uv run python -m obs_sample.client_main --scenario=native # 终端二：客户端 +``` + +样例详细说明见 [observability-sample/README-zh.md](../../a2a-t-sample/observability-sample/README-zh.md)。 + +### 1.7.3 自动记录的内容 + +接入后无需编写任何观测代码，SDK 按交互动作自动记录： + +| 交互动作 | 自动记录的内容 | +| --- | --- | +| 客户端发起请求（SendMessage / SendStreamingMessage 等） | 客户端入口记录，随请求注入调用链标识 | +| 服务端接收并处理请求 | 服务端入口记录，与客户端入口记录串联为同一条调用链 | +| 服务端流式发出事件（状态变更、结果工件、完成等） | 每个事件一条记录，携带事件类型（`streaming.event.kind`）与任务状态（`task.status`） | +| 收发协商消息（Negotiation-T） | 协商记录，携带协商标识（`negotiation.id`）、当前轮次（`negotiation.round`）与协商动作（`negotiation.performative`）；协商达成一致或终止时自动上报收敛轮数指标 `a2at.negotiation.total_rounds` | +| 服务端推送通知 | 推送记录（`SendMessage-pushNotification`） | +| 一次任务请求处理完成 | 耗时指标 `a2at.task.request.duration`（两端分别记录，按属性 `a2at.span.side` 区分客户端/服务端） | +| 通过 `A2ATClient` 等 SDK 门面调用 LLM | 输入/输出 Token 用量自动记录到随后的调用链上 | +| 关键事件（任务状态变更、结果工件等） | logger `a2at.observability` 上的结构化 JSON 日志，自动携带 `trace_id` / `span_id`，可按调用链标识在日志中反查 | + +调用链记录上还会自动携带 A2A-T 业务属性（属性名以 `gen_ai.agent.a2at.` 为前缀），可在后端按属性过滤和统计，常用属性如下： + +| 属性 | 说明 | 来源 | +| --- | --- | --- | +| `extension.name` | 本次交互使用的 A2A-T 扩展类型（Task-T / Negotiation-T / Notification-T / Authorization-T） | 从消息 metadata 的扩展 URI 提取 | +| `task.type` | 任务类型（如“配置下发”） | 按模板章节标题从报文正文提取 | +| `notification.topic` | 通知订阅主题 | 同上 | +| `authorization.policy.operation.type` | 授权策略的操作类型 | 同上 | +| `negotiation.id` / `negotiation.round` | 协商标识与当前轮次 | 从消息 metadata 的 `negotiationContext` 提取 | + +例如在后端按 `extension.name=Task-T` 过滤，可以只看任务下发类交互的调用链；按 `negotiation.id` 查询，可以还原一场多轮协商的完整过程。 + +### 1.7.4 配置项列表 + +可观测性配置项均从操作系统环境变量读取（非 `.env` 文件）： + +| 环境变量 | 默认值 | 说明 | +| --- | --- | --- | +| `A2AT_OBSERVABILITY_ENABLED` | `true` | 总开关；`false` 时模块完全透明 | +| `A2AT_TRACE_ENABLED` / `A2AT_METRIC_ENABLED` / `A2AT_LOG_ENABLED` | `true` | 按数据类型独立开关，例如关闭调用链后指标与日志仍工作 | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | 未设置 | 数据收集端点；未设置时输出到控制台（联调模式） | +| `A2AT_CONSOLE_EXPORTER` | `true` | 置 `false` 关闭联调模式的控制台输出（未配置收集端点时不产生任何输出） | +| `OTEL_SERVICE_NAME`（或 `A2AT_SERVICE_NAME`） | `a2a-t-agent` | 服务名 | +| `A2AT_AGENT_ROLE` | 未设置 | 进程角色标记（如 `client` / `server`），写入资源属性 `agent.role` | +| `A2AT_EXPORTER_PROTOCOL` | `grpc` | 上报协议，可选 `http/protobuf` | +| `A2AT_EXTRACT_REQUEST` / `A2AT_EXTRACT_RESPONSE` | `false` | 是否把请求/响应报文写入调用链记录的属性（JSON 格式，截断规则同报文日志） | +| `A2AT_LOGS_PAYLOAD_ENABLED` | `false` | 是否在结构化日志中输出报文内容 | +| `A2AT_LOGS_PAYLOAD_MAX_LENGTH` | `4096` | 报文内容单条最大长度（字符），超出部分截断并追加 `[truncated]` 标记 | +| `OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED` | `true` | 硬开关；置 `false` 后所有观测逻辑在模块加载时即停用 | + +> 风险提示：开启 `A2AT_EXTRACT_REQUEST` / `A2AT_EXTRACT_RESPONSE` 或 `A2AT_LOGS_PAYLOAD_ENABLED` 后，请求与响应报文（含提示词内容）会进入调用链属性或日志，可能包含敏感信息，仅限联调阶段使用，正式运行时必须关闭；报文较长时还会显著增加数据量，请配合 `A2AT_LOGS_PAYLOAD_MAX_LENGTH` 控制。如需脱敏，可在构造 `A2ATObservabilityConfig` 时传入 `payload_redactor` 自定义脱敏函数（先脱敏、后截断），并通过 `register_client_factory(factory, config=...)` 或 `A2ATRequestHandlerDecorator(handler, config=...)` 传入。 + +### 1.7.5 在业务代码中补充观测数据（可选） + +自动记录之外，SDK 提供三个补充入口，按需使用： + +**1、给当前环节补充业务属性** + +在 `AgentExecutor.execute` 等业务代码中调用 `current_span()` 获取当前环节的视图，追加自定义属性（例如业务单号）： + +```python +from a2a_t.observability import current_span + +span = current_span() +if span is not None: + span.set_attribute("biz.order_id", order_id) +``` + +未接入可观测性或无进行中的调用链时返回 `None`，调用方判空即可。 + +**2、手动记录一段业务处理** + +用 `A2ATSpan` 记录一段业务处理的耗时与属性： + +```python +from a2a_t.observability import A2ATSpan + +with A2ATSpan("query-ne-config") as span: + span.set_attribute("ne.name", ne_name) + query_ne_config(ne_name) +``` + +**3、记录 SDK 门面调用** + +用 `trace_facade` 包装 `A2ATClient` / `A2ATServer` 实例，为提示词生成、报文校验等 SDK 内部处理各记录一个环节（名称形如 `a2at.sdk.client.generate_task_prompt_from_text`）： + +```python +from pathlib import Path +from a2a_t.client.a2at_client import A2ATClient +from a2a_t.observability import trace_facade + +client = trace_facade(A2ATClient(env_path=Path("package_data/.env"))) +``` + +> 以上入口在未安装可观测依赖或总开关关闭时自动失效（不产生任何记录），业务代码无需额外判断。 + +## 1.8 配置项列表 SDK提供的样例配置文件为`env.example`，复制为 `package_data/.env`（或传入显式 `env_path`）。A2A-T Client和A2A-T Server构造时均读取 `.env` 文件（不读取操作系统环境变量），配置项含义如下：