Open in ChatGPTOpen in Claude

Python: manual instrumentation

Pure manual spans with no instrumentation library, just OTel SDK + raw HTTP. Universal template for any provider/language.

Before you start

These examples export to a local OpenTelemetry Collector over OTLP/gRPC. Deploy the collector first. See Code examples → Deploy an OpenTelemetry Collector.

Install

pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-grpc httpx

Environment variables

export OPENAI_API_KEY="sk-..."
export CX_OTEL_ENDPOINT="http://<COLLECTOR_HOST>:4317"

Script

import json, os
import httpx

# --- OTel imports ---
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import SpanKind, StatusCode

# --- OTel setup: configure tracer provider and OTLP exporter ---
CX_ENDPOINT = os.environ.get("CX_OTEL_ENDPOINT", "http://<COLLECTOR_HOST>:4317")
OPENAI_KEY = os.environ.get("OPENAI_API_KEY", "")

resource = Resource.create({"service.name": "manual-genai-demo",
    "cx.application.name": "my-genai-app", "cx.subsystem.name": "my-service"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint=CX_ENDPOINT, insecure=True)))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("genai-manual", "1.0.0")


def to_semconv(messages):
    """Convert OpenAI messages to new semconv JSON parts format."""
    result = []
    for m in messages:
        parts = []
        if m.get("content"): parts.append({"type": "text", "content": m["content"]})
        if m.get("tool_calls"):
            for tc in m["tool_calls"]:
                parts.append({"type": "tool_call", "id": tc["id"],
                    "name": tc["function"]["name"], "arguments": tc["function"]["arguments"]})
        entry = {"role": m["role"], "parts": parts}
        if m.get("tool_call_id"): entry["tool_call_id"] = m["tool_call_id"]
        result.append(entry)
    return json.dumps(result)


# --- Your app logic with manual OTel GenAI spans ---
def chat(messages, model="gpt-4o-mini", max_tokens=200, user=None):
    # OTel: create a GenAI span with required attributes
    with tracer.start_as_current_span(f"chat {model}", kind=SpanKind.CLIENT,
        attributes={"gen_ai.operation.name": "chat", "gen_ai.provider.name": "openai",
            "gen_ai.request.model": model, "gen_ai.request.max_tokens": max_tokens,
            "gen_ai.input.messages": to_semconv(messages),
            "server.address": "api.openai.com", "server.port": 443}) as span:
        if user: span.set_attribute("gen_ai.request.user", user)

        body = {"model": model, "messages": messages, "max_tokens": max_tokens}
        if user: body["user"] = user

        resp = httpx.post("https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {OPENAI_KEY}",
                     "Content-Type": "application/json"}, json=body, timeout=60.0)
        resp.raise_for_status()
        data = resp.json()

        usage = data.get("usage", {})
        choices = data.get("choices", [])
        span.set_attribute("gen_ai.response.model", data.get("model", model))
        span.set_attribute("gen_ai.response.id", data.get("id", ""))
        span.set_attribute("gen_ai.response.finish_reasons",
            json.dumps([c.get("finish_reason") for c in choices]))
        span.set_attribute("gen_ai.usage.input_tokens", usage.get("prompt_tokens", 0))
        span.set_attribute("gen_ai.usage.output_tokens", usage.get("completion_tokens", 0))

        out = []
        for c in choices:
            m = c.get("message", {})
            parts = []
            if m.get("content"): parts.append({"type": "text", "content": m["content"]})
            if m.get("tool_calls"):
                for tc in m["tool_calls"]:
                    parts.append({"type": "tool_call", "id": tc["id"],
                        "name": tc["function"]["name"], "arguments": tc["function"]["arguments"]})
            out.append({"role": m.get("role", "assistant"), "parts": parts})
        span.set_attribute("gen_ai.output.messages", json.dumps(out))
        return data


result = chat(
    messages=[{"role": "system", "content": "You are a concise assistant."},
              {"role": "user", "content": "What is OpenTelemetry in one sentence?"}],
    user="user-42")
print(f"Response: {result['choices'][0]['message']['content']}")

provider.force_flush()
provider.shutdown()

Universal template

To adapt for Anthropic, Gemini, or any other provider, change the HTTP endpoint, request body format, and response parsing. The OTel span attributes remain identical.

Next steps

Look up which open-source library to use for your provider in Compatibility matrix.