diff --git a/src/agents/tracing/processors.py b/src/agents/tracing/processors.py index b61f3e7976..89ee54a533 100644 --- a/src/agents/tracing/processors.py +++ b/src/agents/tracing/processors.py @@ -10,6 +10,7 @@ from collections.abc import Callable from functools import cached_property from typing import Any, cast +from urllib.parse import urlsplit import httpx2 @@ -23,6 +24,51 @@ from .spans import Span from .traces import Trace +# Warn once per process when OPENAI_BASE_URL is set but traces still go to OpenAI. +_warned_default_trace_endpoint_with_custom_model_base = False + + +def _split_url(url: str) -> tuple[str, str, int | None, str] | None: + try: + parts = urlsplit(url) + hostname = parts.hostname or "" + try: + port = parts.port + except ValueError: + port = None + return parts.scheme, hostname, port, parts.path + except ValueError: + return None + + +def _url_origin_and_path(url: str) -> tuple[str, str] | None: + parsed = _split_url(url) + if parsed is None: + return None + scheme, hostname, port, path = parsed + if not hostname: + return None + scheme = scheme.lower() or "https" + hostname = hostname.lower() + if port is None or (scheme == "https" and port == 443) or (scheme == "http" and port == 80): + origin = f"{scheme}://{hostname}" + else: + origin = f"{scheme}://{hostname}:{port}" + return origin, path.rstrip("/") + + +def _url_origin(url: str) -> str | None: + parsed = _url_origin_and_path(url) + return None if parsed is None else parsed[0] + + +def _redact_url_for_log(url: str) -> str: + """Drop userinfo, path, query, and fragment so gateway credentials never reach logs.""" + origin = _url_origin(url) + if origin is None: + return "" + return origin + class ConsoleSpanExporter(TracingExporter): """Prints the traces and spans to the console.""" @@ -59,7 +105,7 @@ def __init__( api_key: str | None = None, organization: str | None = None, project: str | None = None, - endpoint: str = _OPENAI_TRACING_INGEST_ENDPOINT, + endpoint: str | None = None, max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 30.0, @@ -72,7 +118,9 @@ def __init__( `os.environ["OPENAI_ORG_ID"]` if not provided. project: The OpenAI project to use. Defaults to `os.environ["OPENAI_PROJECT_ID"]` if not provided. - endpoint: The HTTP endpoint to which traces/spans are posted. + endpoint: The HTTP endpoint to which traces/spans are posted. Defaults to + `os.environ["OPENAI_TRACING_INGEST_ENDPOINT"]` if not provided, otherwise the + OpenAI traces ingest endpoint. This is independent of `OPENAI_BASE_URL`. max_retries: Maximum number of retries upon failures. base_delay: Base delay (in seconds) for the first backoff. max_delay: Maximum delay (in seconds) for backoff growth. @@ -80,7 +128,7 @@ def __init__( self._api_key = api_key self._organization = organization self._project = project - self.endpoint = endpoint + self._endpoint = endpoint self.max_retries = max_retries self.base_delay = base_delay self.max_delay = max_delay @@ -115,6 +163,54 @@ def organization(self): def project(self): return self._project or os.environ.get("OPENAI_PROJECT_ID") + def _invalidate_endpoint(self) -> None: + self.__dict__.pop("_resolved_endpoint", None) + + @property + def endpoint(self) -> str: + if "_resolved_endpoint" not in self.__dict__: + self.__dict__["_resolved_endpoint"] = ( + self._endpoint + or os.environ.get("OPENAI_TRACING_INGEST_ENDPOINT") + or self._OPENAI_TRACING_INGEST_ENDPOINT + ) + return self.__dict__["_resolved_endpoint"] + + @endpoint.setter + def endpoint(self, value: str) -> None: + self._endpoint = value + self._invalidate_endpoint() + + def _warn_if_trace_endpoint_ignores_model_base_url(self) -> None: + global _warned_default_trace_endpoint_with_custom_model_base + if _warned_default_trace_endpoint_with_custom_model_base: + return + model_base = (os.environ.get("OPENAI_BASE_URL") or "").strip() + if not model_base: + return + if not self._should_sanitize_for_openai_tracing_api(): + return + model_origin = _url_origin(model_base) + if model_origin is not None and model_origin == _url_origin( + self._OPENAI_TRACING_INGEST_ENDPOINT + ): + return + _warned_default_trace_endpoint_with_custom_model_base = True + if self._endpoint is not None: + redirect_hint = ( + "Pass a different endpoint= to BackendSpanExporter, or omit that argument so " + "OPENAI_TRACING_INGEST_ENDPOINT can redirect traces" + ) + else: + redirect_hint = "Set OPENAI_TRACING_INGEST_ENDPOINT to redirect traces" + logger.warning( + "[non-fatal] Tracing still exports to %s while OPENAI_BASE_URL is %s. " + "%s, or disable tracing with OPENAI_AGENTS_DISABLE_TRACING=1.", + _redact_url_for_log(self.endpoint), + _redact_url_for_log(model_base), + redirect_hint, + ) + def export(self, items: list[Trace | Span[Any]]) -> None: self._export_with_deadline(items, deadline=None) @@ -133,6 +229,8 @@ def _export_with_deadline(self, items: list[Trace | Span[Any]], deadline: float logger.warning("OPENAI_API_KEY is not set, skipping trace export") continue + self._warn_if_trace_endpoint_ignores_model_base_url() + sanitize_for_openai = self._should_sanitize_for_openai_tracing_api() data: list[dict[str, Any]] = [] for item in grouped: @@ -256,7 +354,9 @@ def _sleep_before_retry(self, sleep_time: float, deadline: float | None) -> bool return True def _should_sanitize_for_openai_tracing_api(self) -> bool: - return self.endpoint.rstrip("/") == self._OPENAI_TRACING_INGEST_ENDPOINT.rstrip("/") + endpoint = _url_origin_and_path(self.endpoint) + default = _url_origin_and_path(self._OPENAI_TRACING_INGEST_ENDPOINT) + return endpoint is not None and endpoint == default def _sanitize_for_openai_tracing_api(self, payload_item: dict[str, Any]) -> dict[str, Any]: """Drop or truncate span fields known to be rejected by traces ingest.""" diff --git a/tests/tracing/test_processor_endpoint.py b/tests/tracing/test_processor_endpoint.py new file mode 100644 index 0000000000..f34a2c3943 --- /dev/null +++ b/tests/tracing/test_processor_endpoint.py @@ -0,0 +1,377 @@ +from __future__ import annotations + +import logging +from types import SimpleNamespace +from typing import Any, cast + +import agents.tracing.processors as processors +from agents.tracing.processors import BackendSpanExporter, _redact_url_for_log +from agents.tracing.spans import Span +from agents.tracing.traces import Trace + +DEFAULT_ENDPOINT = BackendSpanExporter._OPENAI_TRACING_INGEST_ENDPOINT +CUSTOM_ENDPOINT = "https://traces.example.test/v1/traces/ingest" +MODEL_BASE = "https://gateway.example.test/v1" +MODEL_ORIGIN = "https://gateway.example.test" +DEFAULT_ORIGIN = "https://api.openai.com" + + +def _reset_warning(monkeypatch) -> None: + monkeypatch.setattr(processors, "_warned_default_trace_endpoint_with_custom_model_base", False) + + +def _export_once(monkeypatch, exporter: BackendSpanExporter | None = None) -> BackendSpanExporter: + class DummyItem: + tracing_api_key = None + + def export(self) -> dict[str, str]: + return {"id": "span-1"} + + def fake_post(*, url, headers, json): + return SimpleNamespace(status_code=200, text="ok") + + exporter = exporter or BackendSpanExporter() + exporter.set_api_key("test-key") + monkeypatch.setattr(exporter, "_client", SimpleNamespace(post=fake_post)) + exporter.export(cast(list[Trace | Span[Any]], [DummyItem()])) + return exporter + + +def test_endpoint_defaults_to_openai_ingest(monkeypatch): + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + _reset_warning(monkeypatch) + + exporter = BackendSpanExporter() + + assert exporter.endpoint == DEFAULT_ENDPOINT + + +def test_endpoint_from_env(monkeypatch): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + _reset_warning(monkeypatch) + + exporter = BackendSpanExporter() + + assert exporter.endpoint == CUSTOM_ENDPOINT + + +def test_constructor_endpoint_wins_over_env(monkeypatch): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + _reset_warning(monkeypatch) + + exporter = BackendSpanExporter(endpoint="https://explicit.example.test/ingest") + + assert exporter.endpoint == "https://explicit.example.test/ingest" + + +def test_post_construction_endpoint_assignment_before_first_read(monkeypatch): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + _reset_warning(monkeypatch) + + assigned = "https://assigned.example.test/ingest" + exporter = BackendSpanExporter() + exporter.endpoint = assigned + + assert exporter.endpoint == assigned + assert exporter._endpoint == assigned + assert exporter._should_sanitize_for_openai_tracing_api() is False + + +def test_post_construction_endpoint_assignment_invalidates_cache(monkeypatch): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + _reset_warning(monkeypatch) + + assigned = "https://assigned.example.test/ingest" + exporter = BackendSpanExporter() + assert exporter.endpoint == CUSTOM_ENDPOINT + + exporter.endpoint = assigned + + assert exporter.endpoint == assigned + assert exporter._endpoint == assigned + assert exporter._should_sanitize_for_openai_tracing_api() is False + + +def test_export_posts_to_env_endpoint(monkeypatch): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + _reset_warning(monkeypatch) + + class DummyItem: + tracing_api_key = None + + def export(self) -> dict[str, str]: + return {"id": "span-1"} + + calls: list[dict[str, Any]] = [] + + def fake_post(*, url, headers, json): + calls.append({"url": url, "headers": headers, "json": json}) + return SimpleNamespace(status_code=200, text="ok") + + exporter = BackendSpanExporter() + exporter.set_api_key("test-key") + monkeypatch.setattr(exporter, "_client", SimpleNamespace(post=fake_post)) + exporter.export(cast(list[Trace | Span[Any]], [DummyItem()])) + + assert len(calls) == 1 + assert calls[0]["url"] == CUSTOM_ENDPOINT + + +def test_export_posts_to_assigned_endpoint(monkeypatch): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + _reset_warning(monkeypatch) + + assigned = "https://assigned.example.test/ingest" + + class DummyItem: + tracing_api_key = None + + def export(self) -> dict[str, str]: + return {"id": "span-1"} + + calls: list[dict[str, Any]] = [] + + def fake_post(*, url, headers, json): + calls.append({"url": url, "headers": headers, "json": json}) + return SimpleNamespace(status_code=200, text="ok") + + exporter = BackendSpanExporter() + assert exporter.endpoint == CUSTOM_ENDPOINT + exporter.endpoint = assigned + exporter.set_api_key("test-key") + monkeypatch.setattr(exporter, "_client", SimpleNamespace(post=fake_post)) + exporter.export(cast(list[Trace | Span[Any]], [DummyItem()])) + + assert len(calls) == 1 + assert calls[0]["url"] == assigned + + +def test_constructor_does_not_warn(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", MODEL_BASE) + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + BackendSpanExporter() + BackendSpanExporter() + + assert not [record for record in caplog.records if "Tracing still exports" in record.message] + + +def test_warns_once_when_model_base_url_diverges(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", MODEL_BASE) + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + _export_once(monkeypatch) + + warnings = [ + record.message for record in caplog.records if "Tracing still exports" in record.message + ] + assert len(warnings) == 1 + assert DEFAULT_ORIGIN in warnings[0] + assert MODEL_ORIGIN in warnings[0] + assert "/v1" not in warnings[0] + assert "OPENAI_BASE_URL" in warnings[0] + assert "model traffic uses" not in warnings[0] + assert "OPENAI_TRACING_INGEST_ENDPOINT" in warnings[0] + + +def test_no_warning_until_a_trace_can_be_sent(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", MODEL_BASE) + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + class DummyItem: + tracing_api_key = None + + def export(self) -> dict[str, str]: + return {"id": "span-1"} + + def fake_post(*, url, headers, json): + return SimpleNamespace(status_code=200, text="ok") + + exporter = BackendSpanExporter() + monkeypatch.setattr(exporter, "_client", SimpleNamespace(post=fake_post)) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + exporter.export(cast(list[Trace | Span[Any]], [DummyItem()])) + + assert not [record for record in caplog.records if "Tracing still exports" in record.message] + + exporter.set_api_key("test-key") + with caplog.at_level(logging.WARNING, logger="openai.agents"): + exporter.export(cast(list[Trace | Span[Any]], [DummyItem()])) + + warnings = [ + record.message for record in caplog.records if "Tracing still exports" in record.message + ] + assert len(warnings) == 1 + assert MODEL_ORIGIN in warnings[0] + + +def test_no_warning_when_openai_base_url_is_openai_origin(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", "https://api.openai.com/v1") + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + + assert not [record for record in caplog.records if "Tracing still exports" in record.message] + + +def test_no_warning_when_openai_base_url_has_trailing_slash(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", "https://api.openai.com/v1/") + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + + assert not [record for record in caplog.records if "Tracing still exports" in record.message] + + +def test_export_survives_invalid_model_base_port(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", "https://gateway.example.test:99999/v1") + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + + warnings = [ + record.message for record in caplog.records if "Tracing still exports" in record.message + ] + assert len(warnings) == 1 + assert "99999" not in warnings[0] + assert MODEL_ORIGIN in warnings[0] + assert "/v1" not in warnings[0] + + +def test_no_warning_when_only_openai_api_base_is_set(monkeypatch, caplog): + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + monkeypatch.setenv("OPENAI_API_BASE", MODEL_BASE) + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + + assert not [record for record in caplog.records if "Tracing still exports" in record.message] + + +def test_warning_redacts_credentials_in_logged_urls(monkeypatch, caplog): + secret_base = "https://user:s3cret@gateway.example.test/v1?token=signed" + monkeypatch.setenv("OPENAI_BASE_URL", secret_base) + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + + warnings = [ + record.message for record in caplog.records if "Tracing still exports" in record.message + ] + assert len(warnings) == 1 + assert "user:s3cret" not in warnings[0] + assert "token=signed" not in warnings[0] + assert MODEL_ORIGIN in warnings[0] + assert "/v1" not in warnings[0] + assert _redact_url_for_log(secret_base) in warnings[0] + + +def test_no_warning_when_tracing_endpoint_is_custom(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", MODEL_BASE) + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + + assert not [record for record in caplog.records if "Tracing still exports" in record.message] + + +def test_no_warning_when_tracing_is_disabled(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", MODEL_BASE) + monkeypatch.setenv("OPENAI_AGENTS_DISABLE_TRACING", "1") + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + BackendSpanExporter() + + assert not [record for record in caplog.records if "Tracing still exports" in record.message] + + +def test_warning_redacts_credential_bearing_paths(monkeypatch, caplog): + secret_base = "https://gateway.example.test/v1/tenants/tok_secret" + monkeypatch.setenv("OPENAI_BASE_URL", secret_base) + monkeypatch.delenv("OPENAI_TRACING_INGEST_ENDPOINT", raising=False) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch) + + warnings = [ + record.message for record in caplog.records if "Tracing still exports" in record.message + ] + assert len(warnings) == 1 + assert "tok_secret" not in warnings[0] + assert "/tenants/" not in warnings[0] + assert MODEL_ORIGIN in warnings[0] + + +def test_equivalent_openai_ingest_urls_still_sanitize(monkeypatch): + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + for equivalent in ( + "https://api.openai.com:443/v1/traces/ingest", + "https://API.OPENAI.COM/v1/traces/ingest", + "https://api.openai.com/v1/traces/ingest/", + ): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", equivalent) + exporter = BackendSpanExporter() + assert exporter._should_sanitize_for_openai_tracing_api() is True + + +def test_custom_ingest_endpoint_does_not_sanitize(monkeypatch): + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + exporter = BackendSpanExporter() + assert exporter._should_sanitize_for_openai_tracing_api() is False + + +def test_warning_for_explicit_endpoint_points_at_constructor(monkeypatch, caplog): + monkeypatch.setenv("OPENAI_BASE_URL", MODEL_BASE) + monkeypatch.setenv("OPENAI_TRACING_INGEST_ENDPOINT", CUSTOM_ENDPOINT) + _reset_warning(monkeypatch) + + with caplog.at_level(logging.WARNING, logger="openai.agents"): + _export_once(monkeypatch, BackendSpanExporter(endpoint=DEFAULT_ENDPOINT)) + + warnings = [ + record.message for record in caplog.records if "Tracing still exports" in record.message + ] + assert len(warnings) == 1 + assert "endpoint=" in warnings[0] + assert "omit that argument" in warnings[0] + + +def test_redact_url_for_log_strips_userinfo_path_query_and_fragment(): + assert ( + _redact_url_for_log("https://user:pass@api.example.test:8443/v1/traces?sig=abc#frag") + == "https://api.example.test:8443" + ) + + +def test_redact_url_for_log_handles_invalid_port(): + assert _redact_url_for_log("https://gateway.example.test:99999/v1") == MODEL_ORIGIN