Python SDK

import os
from hue_sdk import Hue

hue = Hue(
    api_key=os.environ["HUE_API_KEY"],
    service_name="addition-demo",
    capture_content=True,  # Record inputs and outputs alongside timing and metadata.
)

try:
    hue.validate_project()  # Verify which project the key belongs to.
    # Keep related requests in the same conversation.
    with hue.context(session_id="demo-session"):
        with hue.span("demo.run") as run:
            run.set_input({"a": 2, "b": 2})
            with hue.tool("add") as tool:
                tool.set_input({"a": 2, "b": 2})
                result = 2 + 2
                tool.set_output(result)
            run.set_output(result)
            print("Trace ID:", run.trace_id)

    # Confirm delivery before the process exits.
    if not hue.force_flush(timeout_millis=30_000):
        print("Export counters:", hue.export_status)
        raise RuntimeError("Hue did not acknowledge all telemetry.")
finally:
    hue.shutdown_safe(timeout_millis=1000)

.venv/bin/python trace.py

if not hue.force_flush():
    raise RuntimeError("Hue did not acknowledge the telemetry.")
confirmation = hue.verify_trace(
    request_trace_id,
    expected_span_ids=[request_span_id],
    required_fields=["model"],
    timeout_millis=10_000,
)
if not confirmation.verified:
    raise RuntimeError("Expected trace evidence did not arrive before the deadline.")
print(confirmation.receipt.trace_url)

import os
from hue_sdk import create_hue_safe

# Initialize after fork, once per serving process.
hue = create_hue_safe(
    api_key=os.environ.get("HUE_API_KEY"),
    service_name="customer-agent",
    capture_content=os.environ.get("HUE_CAPTURE_CONTENT") != "false",  # Content is on unless explicitly disabled.
    enabled=os.environ.get("HUE_TRACING_ENABLED") != "false",
    max_queue_bytes=8 * 1024 * 1024,
)

def handle_request(inputs):
    with hue.span("agent.request") as span:
        span.set_input(inputs)
        result = existing_agent(inputs)
        span.set_output(result)
        return result

# At process shutdown, after stopping new requests:
def shutdown_tracing():
    return hue.shutdown_safe(timeout_millis=1000)
# Register shutdown_tracing in your shutdown hook; observe its result without raising.

import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anyio


@asynccontextmanager
async def lifespan(app):
    try:
        yield
    finally:
        await asyncio.to_thread(hue.shutdown_safe, 1000)


class TracedStreamingResponse(StreamingResponse):
    async def stream_response(self, send):
        try:
            await super().stream_response(send)
        finally:
            with anyio.CancelScope(shield=True):
                try:
                    await self.body_iterator.aclose()
                finally:
                    await asyncio.to_thread(hue.force_flush_safe, 1000)


app = FastAPI(lifespan=lifespan)


@app.post("/chat")
async def chat(body: dict):
    async def reply():
        with hue.context(session_id=body["session_id"], user_id=body["user_id"]), hue.span("support.reply") as request:
            request.set_input(body["question"])
            parts = []
            try:
                async for chunk in stream_agent(body["question"]):  # your async streaming call
                    parts.append(chunk)
                    yield chunk
            finally:
                request.set_output("".join(parts))

    return TracedStreamingResponse(reply())