diff --git a/.github/workflows/openrouter-python-tests.yml b/.github/workflows/openrouter-python-tests.yml new file mode 100644 index 00000000..5958a281 --- /dev/null +++ b/.github/workflows/openrouter-python-tests.yml @@ -0,0 +1,69 @@ +name: OpenRouter Python SDK contracts + +on: + pull_request: + branches: [main, dev] + paths: + - 'python/frameworks/openrouter/**' + - 'python/frameworks/openai/**' + - 'python/fi_instrumentation/**' + - '.github/workflows/openrouter-python-tests.yml' + push: + branches: [dev] + paths: + - 'python/frameworks/openrouter/**' + - 'python/frameworks/openai/**' + - 'python/fi_instrumentation/**' + - '.github/workflows/openrouter-python-tests.yml' + +permissions: + contents: read + +jobs: + sdk-contract: + runs-on: ubuntu-24.04 + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + include: + - python: '3.10' + openrouter: '1.3.0' + artifact: wheel + - python: '3.13' + openrouter: '1.3.107' + artifact: sdist + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 + with: + python-version: ${{ matrix.python }} + - name: Build and install packages + shell: bash + env: + ARTIFACT: ${{ matrix.artifact }} + OPENROUTER_VERSION: ${{ matrix.openrouter }} + run: | + python -m venv .openrouter-test-venv + .openrouter-test-venv/bin/pip install build + .openrouter-test-venv/bin/python -m build "--$ARTIFACT" python/frameworks/openrouter + .openrouter-test-venv/bin/python -m build "--$ARTIFACT" python/frameworks/openai + if [ "$ARTIFACT" = wheel ]; then + packages=(python/frameworks/{openrouter,openai}/dist/*.whl) + else + packages=(python/frameworks/{openrouter,openai}/dist/*.tar.gz) + fi + .openrouter-test-venv/bin/pip install "${packages[@]}" \ + "openrouter==$OPENROUTER_VERSION" pytest pytest-asyncio + - name: Run SDK and attribution tests without external traffic + shell: bash + run: | + sudo unshare --net bash -c ' + set -e + ip link set lo up + "$1" -I -m pytest python/frameworks/openrouter/tests \ + python/frameworks/openai/tests/test_openai_compatible_spans.py \ + -q -p no:cacheprovider + ' _ "$PWD/.openrouter-test-venv/bin/python" diff --git a/.release-please-manifest.json b/.release-please-manifest.json index c87a9bd4..57ea875e 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -129,5 +129,6 @@ "python": "1.2.2", "typescript/packages/traceai_elevenlabs": "0.1.0", "go/bedrock": "0.1.0", - "typescript/packages/traceai_huggingface_transformers": "0.1.0" + "typescript/packages/traceai_huggingface_transformers": "0.1.0", + "python/frameworks/openrouter": "0.1.0" } diff --git a/python/frameworks/openai/tests/test_openai_compatible_spans.py b/python/frameworks/openai/tests/test_openai_compatible_spans.py index 19880502..1f4a37d6 100644 --- a/python/frameworks/openai/tests/test_openai_compatible_spans.py +++ b/python/frameworks/openai/tests/test_openai_compatible_spans.py @@ -199,6 +199,9 @@ async def run(): @pytest.mark.parametrize("base_url,provider", [ ("https://api.openai.com/v1", "openai"), + ("https://openrouter.ai/api/v1", "openrouter"), + ("https://OPENROUTER.AI./api/v1", "openrouter"), + ("https://openrouter.ai.customer.example/api/v1", "openai"), ("https://my-resource.openai.azure.com/openai/v1/", "azure"), ("https://my-resource.services.ai.azure.com/openai/v1/", "azure"), ("https://generativelanguage.googleapis.com/v1beta/openai/", "google"), diff --git a/python/frameworks/openai/traceai_openai/_providers.py b/python/frameworks/openai/traceai_openai/_providers.py index 2c426e46..ef090c2e 100644 --- a/python/frameworks/openai/traceai_openai/_providers.py +++ b/python/frameworks/openai/traceai_openai/_providers.py @@ -15,6 +15,7 @@ # Exact API hosts. _HOSTS = { "api.openai.com": FiLLMProviderValues.OPENAI.value, + "openrouter.ai": "openrouter", "ollama.com": "ollama", "generativelanguage.googleapis.com": FiLLMProviderValues.GOOGLE.value, "api.anthropic.com": FiLLMProviderValues.ANTHROPIC.value, diff --git a/python/frameworks/openrouter/CHANGELOG.md b/python/frameworks/openrouter/CHANGELOG.md new file mode 100644 index 00000000..74487a06 --- /dev/null +++ b/python/frameworks/openrouter/CHANGELOG.md @@ -0,0 +1,5 @@ +# Changelog + +## 0.1.0 + +- Instrument the dedicated OpenRouter Python SDK's chat, embeddings, and Responses methods, including asynchronous requests and stream completion. diff --git a/python/frameworks/openrouter/README.md b/python/frameworks/openrouter/README.md new file mode 100644 index 00000000..a14274de --- /dev/null +++ b/python/frameworks/openrouter/README.md @@ -0,0 +1,32 @@ +# OpenRouter Python SDK tracing + +Trace calls made through the dedicated [`openrouter` Python SDK](https://pypi.org/project/openrouter/) with the maintained `traceAI-openrouter` package. For the OpenAI-compatible client, use `traceAI-openai` instead; do not instrument the same request with both packages or HTTPX tracing. + +```bash +pip install traceAI-openrouter +``` + +Set `FI_API_KEY`, `FI_SECRET_KEY`, and `OPENROUTER_API_KEY` in your environment. Then: + +```python +import os + +from fi_instrumentation import register +from fi_instrumentation.fi_types import ProjectType +from openrouter import OpenRouter +from traceai_openrouter import OpenRouterInstrumentor + +provider = register(project_type=ProjectType.OBSERVE, project_name="openrouter-demo") +OpenRouterInstrumentor().instrument(tracer_provider=provider) +client = OpenRouter(api_key=os.environ["OPENROUTER_API_KEY"]) +answer = client.chat.send( + model="openai/gpt-4o-mini", + messages=[{"role": "user", "content": "What is the capital of Japan?"}], +) +print(answer.choices[0].message.content) +provider.force_flush() +``` + +Find the `openrouter.chat` span in your Future AGI project. The adapter covers synchronous and asynchronous Chat, Embeddings, Responses and deprecated `beta.responses` calls. Streams finish when fully consumed or closed; final usage is available only when the server sends a usage chunk. For early exits, close the stream, or use a context manager. + +The default TraceConfig captures plain-text chat messages, plain-text Responses inputs and textual outputs, capped at 32 KiB. Structured or multimodal input, embedding vectors and content, image payloads, exception messages and stacks, and authorization headers are never included. Pass `TraceConfig(hide_inputs=True, hide_outputs=True)` to `instrument(config=...)` to remove prompt/response content. Directional message, text and image privacy flags also suppress the corresponding generic input/output attributes; hiding input images suppresses the entire input value because a text string can contain an image reference. The package does not send requests to OpenRouter on its own. diff --git a/python/frameworks/openrouter/pyproject.toml b/python/frameworks/openrouter/pyproject.toml new file mode 100644 index 00000000..120a9475 --- /dev/null +++ b/python/frameworks/openrouter/pyproject.toml @@ -0,0 +1,26 @@ +[tool.poetry] +name = "traceAI-openrouter" +version = "0.1.0" +description = "OpenTelemetry instrumentation for the dedicated OpenRouter Python SDK" +authors = ["Future AGI "] +readme = "README.md" +license = "Apache-2.0" +packages = [{ include = "traceai_openrouter" }] +classifiers = ["Intended Audience :: Developers", "License :: OSI Approved :: Apache Software License", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Framework :: OpenTelemetry :: Instrumentations"] + +[tool.poetry.dependencies] +python = ">=3.10,<3.14" +fi-instrumentation-otel = ">=1.1.0" +openrouter = ">=1.3.0,<2" +wrapt = ">=1.15.0" + +[tool.poetry.group.dev.dependencies] +pytest = ">=7" +pytest-asyncio = ">=0.23" + +[tool.poetry.urls] +Repository = "https://github.com/future-agi/traceAI/tree/dev/python/frameworks/openrouter" + +[build-system] +requires = ["poetry-core"] +build-backend = "poetry.core.masonry.api" diff --git a/python/frameworks/openrouter/tests/test_instrumentation.py b/python/frameworks/openrouter/tests/test_instrumentation.py new file mode 100644 index 00000000..4ae8a74e --- /dev/null +++ b/python/frameworks/openrouter/tests/test_instrumentation.py @@ -0,0 +1,216 @@ +"""Real OpenRouter SDK calls with synthetic responses and no external traffic.""" +import asyncio +import json + +import httpx +import pytest +from openrouter import OpenRouter +from fi_instrumentation import TraceConfig +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from traceai_openrouter import OpenRouterInstrumentor + + +@pytest.fixture +def spans(): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + instrumentor = OpenRouterInstrumentor() + instrumentor.instrument(tracer_provider=provider) + yield exporter + instrumentor.uninstrument() + provider.shutdown() + + +def sdk(handler): + transport = httpx.MockTransport(handler) + return OpenRouter(api_key="synthetic-key", client=httpx.Client(transport=transport), + async_client=httpx.AsyncClient(transport=transport)) + + +def chat_body(): + return {"id": "chat-1", "object": "chat.completion", "created": 1, + "model": "vendor/served", "system_fingerprint": None, "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": "Tokyo"}}], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}} + + +def attributes(spans): + finished = spans.get_finished_spans() + assert len(finished) == 1 + return dict(finished[0].attributes) + + +def test_chat_sync_records_model_usage_content_without_secret(spans): + seen = [] + def handle(req): + seen.append(req) + return httpx.Response(200, json=chat_body()) + client = sdk(handle) + result = client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}]) + assert result.choices[0].message.content == "Tokyo" + a = attributes(spans) + assert len(seen) == 1 and seen[0].url.host == "openrouter.ai" + assert a["gen_ai.provider.name"] == "openrouter" + assert a["gen_ai.request.model"] == "vendor/request" + assert a["gen_ai.response.model"] == "vendor/served" + assert a["gen_ai.usage.input_tokens"] == 3 + assert a["gen_ai.usage.output_tokens"] == 2 + assert a["output.value"] == "Tokyo" + assert "synthetic-key" not in str(a) + + +def test_chat_async(spans): + client = sdk(lambda req: httpx.Response(200, json=chat_body())) + result = asyncio.run(client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}])) + assert result.choices[0].message.content == "Tokyo" + assert attributes(spans)["gen_ai.usage.input_tokens"] == 3 + + +def test_error_does_not_export_response_or_key(spans): + client = sdk(lambda req: httpx.Response(401, json={"error": {"message": "synthetic-key is invalid"}})) + with pytest.raises(Exception): + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}]) + a = attributes(spans) + assert a["error.type"] + assert "synthetic-key" not in str(a) + + +def test_chat_stream_finishes_after_consumption_with_final_usage(spans): + chunks = [{"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}]} + for text in ("Tok", "yo")] + chunks.append({"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}}) + body = "".join("data: " + json.dumps(chunk) + "\n\n" for chunk in chunks) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + stream = client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) + assert not spans.get_finished_spans() + assert list(stream) + a = attributes(spans) + assert a["output.value"] == "Tokyo" + assert a["gen_ai.usage.input_tokens"] == 3 + + +def test_embedding_has_usage_and_no_vector_content(spans): + client = sdk(lambda req: httpx.Response(200, json={ + "object": "list", "model": "vendor/embed", "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}], + "usage": {"prompt_tokens": 4, "total_tokens": 4}})) + answer = client.embeddings.generate(model="vendor/embed", input="secret input") + assert len(answer.data[0].embedding) == 2 + a = attributes(spans) + assert a["gen_ai.operation.name"] == "embeddings" + assert a["gen_ai.usage.input_tokens"] == 4 + assert "input.value" not in a and "output.value" not in a + + +def test_responses_and_beta_alias(spans): + body = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": 2, + "error": None, "frequency_penalty": None, "incomplete_details": None, + "instructions": None, "metadata": None, "model": "vendor/served", + "output": [], "output_text": "Tokyo", "parallel_tool_calls": False, + "presence_penalty": None, "status": "completed", "temperature": None, + "tool_choice": "auto", "tools": [], "top_p": None, + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}}} + client = sdk(lambda req: httpx.Response(200, json=body)) + assert client.responses.send(model="vendor/request", input="Japan?").output_text == "Tokyo" + assert client.beta.responses.send(model="vendor/request", input="Japan?").output_text == "Tokyo" + finished = spans.get_finished_spans() + assert len(finished) == 2 + assert all(s.attributes["gen_ai.provider.name"] == "openrouter" for s in finished) + assert all(s.attributes.get("gen_ai.usage.output_tokens") == 2 for s in finished), [dict(s.attributes) for s in finished] + + +def test_responses_stream_final_event_usage(spans): + response = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": 2, + "error": None, "frequency_penalty": None, "incomplete_details": None, + "instructions": None, "metadata": None, "model": "vendor/served", + "output": [], "output_text": "Tokyo", "parallel_tool_calls": False, + "presence_penalty": None, "status": "completed", "temperature": None, + "tool_choice": "auto", "tools": [], "top_p": None, + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}}} + event = {"type": "response.completed", "sequence_number": 1, "response": response} + payload = "data: " + json.dumps(event) + "\n\ndata: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=payload.encode())) + stream = client.responses.send(model="vendor/request", input="Japan?", stream=True) + assert list(stream) + a = attributes(spans) + assert a["gen_ai.response.model"] == "vendor/served" + assert a["gen_ai.usage.output_tokens"] == 2 + assert a["output.value"] == "Tokyo" + + +def test_stream_early_close_marks_incomplete_without_claiming_final_text(spans): + chunks = [{"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tok"}, "finish_reason": None}]}] + body = "".join("data: " + json.dumps(chunk) + "\n\n" for chunk in chunks) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + with client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) as stream: + next(stream) + a = attributes(spans) + assert a["openrouter.stream.incomplete"] is True + assert "output.value" not in a + + +def test_chat_async_stream_usage(spans): + chunks = [{"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tokyo"}, "finish_reason": None}]}, + {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}}] + body = "".join("data: " + json.dumps(chunk) + "\n\n" for chunk in chunks) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) + return [chunk async for chunk in stream] + assert len(asyncio.run(run())) == 2 + a = attributes(spans) + assert a["output.value"] == "Tokyo" + assert a["gen_ai.usage.input_tokens"] == 3 + + +def test_chat_async_close_ends_incomplete_span(spans): + chunk = {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tok"}, "finish_reason": None}]} + body = "data: " + json.dumps(chunk) + "\n\ndata: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) + await stream.__anext__() + await stream.close() + asyncio.run(run()) + assert attributes(spans)["openrouter.stream.incomplete"] is True + + +def test_hide_inputs_outputs(spans): + OpenRouterInstrumentor().uninstrument() + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + instrumentor = OpenRouterInstrumentor() + instrumentor.instrument(tracer_provider=provider, config=TraceConfig(hide_inputs=True, hide_outputs=True)) + try: + client = sdk(lambda req: httpx.Response(200, json=chat_body())) + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "secret prompt"}]) + a = attributes(exporter) + assert "input.value" not in a and "output.value" not in a + assert a["gen_ai.usage.input_tokens"] == 3 + finally: + instrumentor.uninstrument() + provider.shutdown() + + +def test_uninstrument_restores_methods(spans): + instrumentor = OpenRouterInstrumentor() + instrumentor.uninstrument() + client = sdk(lambda req: httpx.Response(200, json=chat_body())) + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}]) + assert spans.get_finished_spans() == () diff --git a/python/frameworks/openrouter/tests/test_review_regressions.py b/python/frameworks/openrouter/tests/test_review_regressions.py new file mode 100644 index 00000000..0ffbb9c2 --- /dev/null +++ b/python/frameworks/openrouter/tests/test_review_regressions.py @@ -0,0 +1,207 @@ +"""Privacy, in-band failures, and nested request context using the real SDK.""" +import asyncio +import json + +import httpx +import pytest +from fi_instrumentation import TraceConfig +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode + +from traceai_openrouter import OpenRouterInstrumentor +from test_instrumentation import chat_body, sdk + + +@pytest.fixture +def instrument(): + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + inst = OpenRouterInstrumentor() + def start(config=None): + inst.instrument(tracer_provider=provider, config=config or TraceConfig()) + return provider, exporter + yield start + inst.uninstrument() + provider.shutdown() + + +@pytest.mark.parametrize("flag,content", [ + ("hide_input_messages", "SYNTHETIC_PRIVATE_MESSAGE"), + ("hide_input_text", "SYNTHETIC_PRIVATE_MESSAGE"), + ("hide_input_images", "data:image/png;base64,SYNTHETIC_IMAGE_MARKER"), +]) +def test_chat_input_privacy(instrument, flag, content): + _, exporter = instrument(TraceConfig(**{flag: True})) + messages = [{"role": "user", "content": [ + {"type": "text", "text": "SYNTHETIC_PRIVATE_MESSAGE"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,SYNTHETIC_IMAGE_MARKER"}}, + ]}] + sdk(lambda req: httpx.Response(200, json=chat_body())).chat.send(model="vendor/request", messages=messages) + attrs = dict(exporter.get_finished_spans()[0].attributes) + assert content not in str(attrs) + assert "input.value" not in attrs + + +@pytest.mark.parametrize("flag", ["hide_output_messages", "hide_output_text"]) +def test_chat_output_privacy(instrument, flag): + _, exporter = instrument(TraceConfig(**{flag: True})) + sdk(lambda req: httpx.Response(200, json=chat_body())).chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}]) + attrs = dict(exporter.get_finished_spans()[0].attributes) + assert "Tokyo" not in str(attrs) + assert "output.value" not in attrs + + +def chat_error_stream(): + chunks = [ + {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tok"}, "finish_reason": None}]}, + {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [], "error": {"code": 502, "message": "SYNTHETIC_PRIVATE_ERROR"}}, + ] + return httpx.Response(200, headers={"content-type": "text/event-stream"}, + content=("".join("data: " + json.dumps(c) + "\n\n" for c in chunks) + "data: [DONE]\n\n").encode()) + + +@pytest.mark.parametrize("async_call", [False, True]) +def test_in_band_stream_error_drops_partial_answer(instrument, async_call): + _, exporter = instrument() + client = sdk(lambda req: chat_error_stream()) + if async_call: + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True) + return [part async for part in stream] + chunks = asyncio.run(run()) + else: + chunks = list(client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True)) + assert len(chunks) == 2 + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert "output.value" not in span.attributes + assert "SYNTHETIC_PRIVATE_ERROR" not in str(dict(span.attributes)) + + +@pytest.mark.parametrize("state", ["failed", "incomplete"]) +def test_responses_failure_drops_partial_answer(instrument, state): + _, exporter = instrument() + body = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": 2, + "error": {"code": "server_error", "message": "SYNTHETIC_PRIVATE_ERROR"} if state == "failed" else None, + "frequency_penalty": None, "incomplete_details": None, "instructions": None, + "metadata": None, "model": "vendor/served", "output": [], "output_text": "partial", + "parallel_tool_calls": False, "presence_penalty": None, "status": state, + "temperature": None, "tool_choice": "auto", "tools": [], "top_p": None, + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, "output_tokens_details": {"reasoning_tokens": 0}}} + result = sdk(lambda req: httpx.Response(200, json=body)).responses.send(model="vendor/request", input="hi") + assert result.status == state + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert "output.value" not in span.attributes + assert "SYNTHETIC_PRIVATE_ERROR" not in str(dict(span.attributes)) + + +@pytest.mark.parametrize("async_call", [False, True]) +@pytest.mark.parametrize("beta", [False, True]) +def test_responses_stream_error_event_drops_partial_answer(instrument, async_call, beta): + _, exporter = instrument() + events = [ + {"type": "response.output_text.delta", "sequence_number": 1, "item_id": "item_1", + "output_index": 0, "content_index": 0, "logprobs": [], "delta": "partial"}, + {"type": "error", "sequence_number": 2, "code": "server_error", + "message": "SYNTHETIC_PRIVATE_ERROR", "param": None}, + ] + payload = "".join("data: " + json.dumps(event) + "\n\n" for event in events) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=payload.encode())) + responses = client.beta.responses if beta else client.responses + if async_call: + async def run(): + stream = await responses.send_async(model="vendor/request", input="hi", stream=True) + return [event async for event in stream] + chunks = asyncio.run(run()) + else: + chunks = list(responses.send(model="vendor/request", input="hi", stream=True)) + assert [chunk.type for chunk in chunks] == ["response.output_text.delta", "error"] + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert "output.value" not in span.attributes + assert "SYNTHETIC_PRIVATE_ERROR" not in str(dict(span.attributes)) + + +@pytest.mark.parametrize("async_call", [False, True]) +@pytest.mark.parametrize("beta", [False, True]) +def test_cancelled_responses_drops_partial_answer(instrument, async_call, beta): + _, exporter = instrument() + body = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": None, + "error": None, "frequency_penalty": None, "incomplete_details": None, + "instructions": None, "metadata": None, "model": "vendor/served", "output": [], + "output_text": "partial", "parallel_tool_calls": False, "presence_penalty": None, + "status": "cancelled", "temperature": None, "tool_choice": "auto", "tools": [], + "top_p": None, "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}}} + client = sdk(lambda req: httpx.Response(200, json=body)) + responses = client.beta.responses if beta else client.responses + if async_call: + result = asyncio.run(responses.send_async(model="vendor/request", input="hi")) + else: + result = responses.send(model="vendor/request", input="hi") + assert result.status == "cancelled" + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert span.attributes["gen_ai.response.model"] == "vendor/served" + assert span.attributes["gen_ai.usage.input_tokens"] == 3 + assert "output.value" not in span.attributes + + +@pytest.mark.parametrize("async_call", [False, True]) +def test_http_child_nested_under_llm(instrument, async_call): + provider, exporter = instrument() + tracer = provider.get_tracer("test") + def handler(req): + with tracer.start_as_current_span("http.client"): + return httpx.Response(200, json=chat_body()) + client = sdk(handler) + with tracer.start_as_current_span("caller"): + if async_call: + asyncio.run(client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "hi"}])) + else: + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}]) + spans = {span.name: span for span in exporter.get_finished_spans()} + assert spans["openrouter.chat"].parent.span_id == spans["caller"].context.span_id + assert spans["http.client"].parent.span_id == spans["openrouter.chat"].context.span_id + + +def test_multimodal_input_never_serializes_image_by_default(instrument): + _, exporter = instrument() + messages = [{"role": "user", "content": [ + {"type": "text", "text": "hello"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,SYNTHETIC_IMAGE_MARKER"}}, + ]}] + sdk(lambda req: httpx.Response(200, json=chat_body())).chat.send(model="vendor/request", messages=messages) + assert "input.value" not in exporter.get_finished_spans()[0].attributes + + +@pytest.mark.parametrize("async_call", [False, True]) +def test_stream_http_child_nested_under_llm(instrument, async_call): + provider, exporter = instrument() + tracer = provider.get_tracer("test") + chunk = {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tokyo"}, "finish_reason": None}]} + payload = "data: " + json.dumps(chunk) + "\n\ndata: [DONE]\n\n" + def handler(req): + with tracer.start_as_current_span("http.client"): + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=payload.encode()) + client = sdk(handler) + with tracer.start_as_current_span("caller"): + if async_call: + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True) + return [part async for part in stream] + asyncio.run(run()) + else: + list(client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True)) + spans = {span.name: span for span in exporter.get_finished_spans()} + assert spans["http.client"].parent.span_id == spans["openrouter.chat"].context.span_id + assert spans["openrouter.chat"].parent.span_id == spans["caller"].context.span_id diff --git a/python/frameworks/openrouter/traceai_openrouter/__init__.py b/python/frameworks/openrouter/traceai_openrouter/__init__.py new file mode 100644 index 00000000..ef653184 --- /dev/null +++ b/python/frameworks/openrouter/traceai_openrouter/__init__.py @@ -0,0 +1,316 @@ +"""Trace the dedicated Speakeasy-generated ``openrouter`` Python SDK. + +Instrument only inference endpoints. Do not instrument HTTPX or OpenAI alongside +this instrumentor for the same request; those paths would create duplicate spans. +""" +from __future__ import annotations + +import json +from importlib import import_module +from typing import Any, Collection + +from fi_instrumentation import FITracer, TraceConfig +from fi_instrumentation.fi_types import FiSpanKindValues, SpanAttributes as A +from opentelemetry import trace +from opentelemetry.instrumentation.instrumentor import BaseInstrumentor +from opentelemetry.trace import Status, StatusCode +from wrapt import wrap_function_wrapper + +from .version import __version__ + +_ENDPOINTS = ( + ("openrouter.chat", "Chat", "chat", "send"), + ("openrouter.embeddings", "Embeddings", "embeddings", "generate"), + ("openrouter.responses", "Responses", "responses", "send"), + ("openrouter.beta_responses", "BetaResponses", "responses", "send"), +) +_LIMIT = 32768 + + +def _field(value: Any, name: str) -> Any: + if isinstance(value, dict): + return value.get(name) + return getattr(value, name, None) + + +def _text(value: Any) -> str: + """Capture bounded textual output only, never embedding vectors or images.""" + choices = _field(value, "choices") + if choices: + first = choices[0] + msg = _field(first, "delta") or _field(first, "message") + content = _field(msg, "content") + return content[:_LIMIT] if isinstance(content, str) else "" + output_text = _field(value, "output_text") + if isinstance(output_text, str) and output_text: + return output_text[:_LIMIT] + output = _field(value, "output") + if output: + parts = [] + for item in output: + for block in _field(item, "content") or []: + text = _field(block, "text") + if isinstance(text, str): + parts.append(text) + return "".join(parts)[:_LIMIT] + return "" + + +def _usage(span: Any, value: Any) -> None: + usage = _field(value, "usage") + if not usage: + return + for attr, names in ( + (A.GEN_AI_USAGE_INPUT_TOKENS, ("prompt_tokens", "input_tokens")), + (A.GEN_AI_USAGE_OUTPUT_TOKENS, ("completion_tokens", "output_tokens")), + ): + for name in names: + number = _field(usage, name) + if isinstance(number, int) and not isinstance(number, bool) and number >= 0: + span.set_attribute(attr, number) + break + + +def _result(span: Any, value: Any, content: bool) -> None: + model = _field(value, "model") + if isinstance(model, str): + span.set_attribute(A.GEN_AI_RESPONSE_MODEL, model) + _usage(span, value) + if _field(value, "status") in ("failed", "incomplete", "cancelled"): + # A 200 Responses body can still represent an aborted generation. + # Never export its error message or partial output. + span.set_status(Status(StatusCode.ERROR)) + return + if content: + text = _text(value) + if text: + span.set_attribute(A.OUTPUT_VALUE, text) + + +def _input(span: Any, operation: str, kwargs: dict[str, Any], content: bool) -> None: + if not content: + return + value = kwargs.get("messages") if operation == "chat" else kwargs.get("input") + if value is None or operation == "embeddings": + return # Embedding input can contain arbitrary binary/image payloads. + # Never serialize multimodal objects wholesale into input.value: the generic + # attribute cannot distinguish text from an inline image, URL or tool data. + # Keep only plain chat messages and plain text Responses inputs. + if operation == "chat": + if not isinstance(value, (list, tuple)) or any( + not isinstance(msg, dict) + or not isinstance(msg.get("role"), str) + or not isinstance(msg.get("content"), str) + for msg in value + ): + return + value = [{"role": msg.get("role"), "content": msg["content"]} for msg in value] + elif not isinstance(value, str): + return + if "data:image/" in str(value).lower(): + return + serialized = json.dumps(value) + span.set_attribute(A.INPUT_VALUE, serialized[:_LIMIT]) + span.set_attribute(A.INPUT_MIME_TYPE, "application/json") + + +def _failure(span: Any, exc: BaseException) -> None: + # Exception text and stack traces may contain the API key, request body or URL. + span.set_attribute("error.type", type(exc).__name__) + span.set_status(Status(StatusCode.ERROR)) + span.end() + + +class _Stream: + def __init__(self, source: Any, span: Any, content: bool): + self._source, self._span, self._content = source, span, content + self._ended = False + self._failed = False + self._pieces: list[str] = [] + self._length = 0 + + def __getattr__(self, name: str) -> Any: + return getattr(self._source, name) + + def _observe(self, chunk: Any) -> None: + # Responses SSE events carry final usage on response.completed, not on + # the event itself. The final response supersedes individual deltas. + event_type = _field(chunk, "type") + completed = _field(chunk, "response") + if _field(chunk, "error") is not None or event_type in ("error", "response.failed", "response.incomplete"): + self._failed = True + self._pieces.clear() + self._span.set_status(Status(StatusCode.ERROR)) + if event_type in ("response.completed", "response.failed", "response.incomplete") and completed: + _result(self._span, completed, False) + if _field(completed, "status") in ("failed", "incomplete", "cancelled"): + self._failed = True + self._pieces.clear() + if not self._failed and self._content: + text = _text(completed) + if text: + self._pieces = [text[:_LIMIT]] + self._length = len(self._pieces[0]) + return + _result(self._span, chunk, False) + if not self._failed and self._content and self._length < _LIMIT: + text = _field(chunk, "delta") if event_type == "response.output_text.delta" else _text(chunk) + if isinstance(text, str): + text = text[: _LIMIT - self._length] + self._pieces.append(text) + self._length += len(text) + + def _finish(self, complete: bool = True) -> None: + if not self._ended: + self._ended = True + if not complete: + self._span.set_attribute("openrouter.stream.incomplete", True) + if self._pieces and complete and not self._failed: + self._span.set_attribute(A.OUTPUT_VALUE, "".join(self._pieces)) + self._span.end() + + def __iter__(self): + return self + + def __next__(self): + try: + with trace.use_span(self._span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + chunk = next(self._source) + except StopIteration: + self._finish() + raise + except BaseException as exc: + self._ended = True + _failure(self._span, exc) + raise + self._observe(chunk) + return chunk + + def __aiter__(self): + return self + + async def __anext__(self): + try: + with trace.use_span(self._span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + chunk = await self._source.__anext__() + except StopAsyncIteration: + self._finish() + raise + except BaseException as exc: + self._ended = True + _failure(self._span, exc) + raise + self._observe(chunk) + return chunk + + def close(self): + try: + return self._source.close() + finally: + self._finish(complete=False) + + def __enter__(self): + self._source.__enter__() + return self + + def __exit__(self, *args): + try: + return self._source.__exit__(*args) + finally: + self._finish(complete=False) + + async def __aenter__(self): + await self._source.__aenter__() + return self + + async def __aexit__(self, *args): + try: + return await self._source.__aexit__(*args) + finally: + self._finish(complete=False) + + +class _AsyncStream(_Stream): + async def close(self): + try: + return await self._source.close() + finally: + self._finish(complete=False) + + +def _start(tracer: Any, operation: str, kwargs: dict[str, Any], config: TraceConfig): + kind = FiSpanKindValues.EMBEDDING if operation == "embeddings" else FiSpanKindValues.LLM + model = kwargs.get("model") + attrs = {A.GEN_AI_PROVIDER_NAME: "openrouter", A.GEN_AI_OPERATION_NAME: operation} + if isinstance(model, str): + attrs[A.GEN_AI_REQUEST_MODEL] = model + span = tracer.start_span("openrouter." + operation, attributes=attrs, fi_span_kind=kind) + # Generic input.value bypasses message/image redaction, so omit it for + # every directional privacy setting (including images embedded in strings). + capture_input = not ( + config.hide_inputs or config.hide_input_text or config.hide_input_messages or config.hide_input_images + ) + _input(span, operation, kwargs, capture_input) + return span + + +def _wrap(tracer: Any, operation: str, config: TraceConfig, is_async: bool): + def complete(span, result): + capture_output = not (config.hide_outputs or config.hide_output_text or config.hide_output_messages) + if hasattr(result, "__anext__"): + return _AsyncStream(result, span, capture_output) + if hasattr(result, "__next__"): + return _Stream(result, span, capture_output) + _result(span, result, capture_output) + span.end() + return result + + if is_async: + async def wrapped(wrapped, instance, args, kwargs): + span = _start(tracer, operation, kwargs, config) + try: + with trace.use_span(span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + result = await wrapped(*args, **kwargs) + return complete(span, result) + except BaseException as exc: + _failure(span, exc) + raise + else: + def wrapped(wrapped, instance, args, kwargs): + span = _start(tracer, operation, kwargs, config) + try: + with trace.use_span(span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + result = wrapped(*args, **kwargs) + return complete(span, result) + except BaseException as exc: + _failure(span, exc) + raise + return wrapped + + +class OpenRouterInstrumentor(BaseInstrumentor): # type: ignore[misc] + """Instrument dedicated OpenRouter chat, embeddings and Responses calls.""" + + def instrumentation_dependencies(self) -> Collection[str]: + return ("openrouter >= 1.3.0, < 2",) + + def _instrument(self, **kwargs: Any) -> None: + config = kwargs.get("config") or TraceConfig() + if not isinstance(config, TraceConfig): + raise TypeError("config must be a TraceConfig") + tracer = FITracer(trace.get_tracer(__name__, __version__, kwargs.get("tracer_provider")), config) + self._originals = [] + for module, cls_name, operation, method in _ENDPOINTS: + cls = getattr(import_module(module), cls_name) + for name, is_async in ((method, False), (method + "_async", True)): + original = cls.__dict__[name] + wrap_function_wrapper(module, cls_name + "." + name, _wrap(tracer, operation, config, is_async)) + self._originals.append((cls, name, original)) + + def _uninstrument(self, **kwargs: Any) -> None: + for cls, name, original in getattr(self, "_originals", []): + setattr(cls, name, original) + self._originals = [] + + +__all__ = ["OpenRouterInstrumentor", "__version__"] diff --git a/python/frameworks/openrouter/traceai_openrouter/version.py b/python/frameworks/openrouter/traceai_openrouter/version.py new file mode 100644 index 00000000..75716eb9 --- /dev/null +++ b/python/frameworks/openrouter/traceai_openrouter/version.py @@ -0,0 +1 @@ +__version__ = "0.1.0" # x-release-please-version diff --git a/release-please-config.json b/release-please-config.json index 01c08889..17de55ed 100644 --- a/release-please-config.json +++ b/release-please-config.json @@ -1100,6 +1100,14 @@ "extra-files": [ "src/version.ts" ] + }, + "python/frameworks/openrouter": { + "release-type": "python", + "component": "traceai-openrouter", + "package-name": "traceAI-openrouter", + "extra-files": [ + "traceai_openrouter/version.py" + ] } } }