# 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](https://coralogix.com/docs/user-guides/ai/otel-integration/code-examples/#prerequisites-deploy-an-opentelemetry-collector).

## Install

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

## Environment variables

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

## Script

```python
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](https://coralogix.com/docs/user-guides/ai/otel-integration/providers/index.md).