---
title: Python, manual instrumentation
description: Build GenAI spans by hand in Python with the OpenTelemetry SDK and raw HTTP, a universal template for any provider or language.
url: https://coralogix.com/docs/user-guides/ai/otel-integration/code-examples/python-manual/
product: ai
feature: AI App Catalog
feature_id: ai-app-catalog
feature_identifier: ai/ai-app-catalog
last_updated: '2026-09-03'
---

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