Observability

The Health Universe A2A SDK includes built-in observability. When your agent runs in production on Health Universe, every execution is automatically traced — including LLM calls, document operations, and inter-agent communication — with zero code changes required.

Traces flow into the Health Universe Observer dashboard, giving you and your workspace admins visibility into safety scores, latency, token usage, costs, and user feedback.

What Gets Traced

The SDK automatically traces the following operations:

Agent Execution

Every call to process_message() creates a top-level trace with:

• Agent name and type (AsyncAgent or SubAgent)

• Input (user message) and output (agent response)

• Duration of the entire execution

• Metadata: job ID, user ID, thread ID, app ID

LLM Calls

All calls to OpenAI, Anthropic, and other supported LLM providers are captured automatically via OpenTelemetry instrumentation. Each LLM call records:

• Model name

• Token usage (input, output, total)

• Latency

• Request and response content (when content capture is enabled)

Supported LLM providers: OpenAI, Anthropic, Google Generative AI, Cohere, Hugging Face Transformers

Document Operations

Every document operation is traced as a child span:

Operation	Span Name	Metadata
List documents	document.list	Role filter (source, artifact, all)
Download document	document.download	Document ID
Download extracted text	document.download_extracted	Document ID
Write document	document.write	Document name, filename, content size, storage path

Inter-Agent Communication

Calls to other agents via call_agent() or call_other_agent() are traced as child spans:

Operation	Span Name	Metadata
Call another agent	inter_agent.call:{target_agent}	Target agent identifier, message preview

Adding Custom Traces

While the SDK traces all built-in operations automatically, you can add your own traces for custom logic.

Using the @observe Decorator

from health_universe_a2a import observe

class MyAgent(Agent):
    @observe(capture_input=True, capture_output=True)
    async def analyze_data(self, data: dict) -> dict:
        # This method is automatically traced as a child span
        results = await self.run_analysis(data)
        return results

    async def process_message(self, message: str, context: AgentContext) -> str:
        data = await self.load_data(context)
        results = await self.analyze_data(data)
        return json.dumps(results)

Using the Context Manager

For inline spans within a method:

from health_universe_a2a.observability import traced_span

async def process_message(self, message: str, context: AgentContext) -> str:
    with traced_span("validation", metadata={"input_length": len(message)}) as span:
        validated = self.validate_input(message)
        span.update(output="Validation passed")

    with traced_span("analysis", metadata={"model": "gpt-4o"}) as span:
        result = await self.run_analysis(validated)
        span.update(output=f"Processed {len(result)} items")

    return json.dumps(result)

Both the decorator and context manager are no-ops when observability is disabled — no need for conditional checks in your code.

Viewing Traces

Once your agent is deployed, traces appear in the Health Universe Observer dashboard, which aggregates them into safety scores, usage metrics, cost breakdowns, and more.