REDROOM

Navigation

Breadcrumbs

/ / opt / hc_python / lib64 / python3.12 / site-packages / sentry_sdk / integrations / openai_agents

Create

Create File | Create Folder | Upload Files | Run Command

Edit File Size: 7.22 KB

//opt/hc_python/lib64/python3.12/site-packages/sentry_sdk/integrations/openai_agents/utils.py

Text Base64

import json
from typing import TYPE_CHECKING

import sentry_sdk
from sentry_sdk.ai._openai_completions_api import _transform_system_instructions
from sentry_sdk.ai._openai_responses_api import (
    _get_system_instructions,
    _is_system_instruction,
)
from sentry_sdk.ai.utils import (
    GEN_AI_ALLOWED_MESSAGE_ROLES,
    normalize_message_role,
    normalize_message_roles,
    set_data_normalized,
    truncate_and_annotate_messages,
)
from sentry_sdk.consts import SPANDATA
from sentry_sdk.integrations import DidNotEnable
from sentry_sdk.scope import should_send_default_pii
from sentry_sdk.utils import event_from_exception, safe_serialize

if TYPE_CHECKING:
    from typing import Any

from agents import TResponseInputItem, Usage

from sentry_sdk._types import TextPart

try:
    import agents

except ImportError:
    raise DidNotEnable("OpenAI Agents not installed")

def _capture_exception(exc: "Any") -> None:
    event, hint = event_from_exception(
        exc,
        client_options=sentry_sdk.get_client().options,
        mechanism={"type": "openai_agents", "handled": False},
    )
    sentry_sdk.capture_event(event, hint=hint)

def _set_agent_data(span: "sentry_sdk.tracing.Span", agent: "agents.Agent") -> None:
    span.set_data(
        SPANDATA.GEN_AI_SYSTEM, "openai"
    )  # See footnote for  https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/#gen-ai-system for explanation why.

span.set_data(SPANDATA.GEN_AI_AGENT_NAME, agent.name)

if agent.model_settings.max_tokens:
        span.set_data(
            SPANDATA.GEN_AI_REQUEST_MAX_TOKENS, agent.model_settings.max_tokens
        )

# Get model name from agent.model or fall back to request model (for when agent.model is None/default)
    model_name = None
    if agent.model:
        model_name = agent.model.model if hasattr(agent.model, "model") else agent.model
    elif hasattr(agent, "_sentry_request_model"):
        model_name = agent._sentry_request_model

if model_name:
        span.set_data(SPANDATA.GEN_AI_REQUEST_MODEL, model_name)

if agent.model_settings.presence_penalty:
        span.set_data(
            SPANDATA.GEN_AI_REQUEST_PRESENCE_PENALTY,
            agent.model_settings.presence_penalty,
        )

if agent.model_settings.temperature:
        span.set_data(
            SPANDATA.GEN_AI_REQUEST_TEMPERATURE, agent.model_settings.temperature
        )

if agent.model_settings.top_p:
        span.set_data(SPANDATA.GEN_AI_REQUEST_TOP_P, agent.model_settings.top_p)

if agent.model_settings.frequency_penalty:
        span.set_data(
            SPANDATA.GEN_AI_REQUEST_FREQUENCY_PENALTY,
            agent.model_settings.frequency_penalty,
        )

if len(agent.tools) > 0:
        span.set_data(
            SPANDATA.GEN_AI_REQUEST_AVAILABLE_TOOLS,
            safe_serialize([vars(tool) for tool in agent.tools]),
        )

def _set_usage_data(span: "sentry_sdk.tracing.Span", usage: "Usage") -> None:
    span.set_data(SPANDATA.GEN_AI_USAGE_INPUT_TOKENS, usage.input_tokens)
    span.set_data(
        SPANDATA.GEN_AI_USAGE_INPUT_TOKENS_CACHED,
        usage.input_tokens_details.cached_tokens,
    )
    span.set_data(SPANDATA.GEN_AI_USAGE_OUTPUT_TOKENS, usage.output_tokens)
    span.set_data(
        SPANDATA.GEN_AI_USAGE_OUTPUT_TOKENS_REASONING,
        usage.output_tokens_details.reasoning_tokens,
    )
    span.set_data(SPANDATA.GEN_AI_USAGE_TOTAL_TOKENS, usage.total_tokens)

def _set_input_data(
    span: "sentry_sdk.tracing.Span", get_response_kwargs: "dict[str, Any]"
) -> None:
    if not should_send_default_pii():
        return
    request_messages = []

messages: "str | list[TResponseInputItem]" = get_response_kwargs.get("input", [])

instructions_text_parts: "list[TextPart]" = []
    explicit_instructions = get_response_kwargs.get("system_instructions")
    if explicit_instructions is not None:
        instructions_text_parts.append(
            {
                "type": "text",
                "content": explicit_instructions,
            }
        )

system_instructions = _get_system_instructions(messages)

# Deliberate use of function accepting completions API type because
    # of shared structure FOR THIS PURPOSE ONLY.
    instructions_text_parts += _transform_system_instructions(system_instructions)

if len(instructions_text_parts) > 0:
        span.set_data(
            SPANDATA.GEN_AI_SYSTEM_INSTRUCTIONS,
            json.dumps(instructions_text_parts),
        )

non_system_messages = [
        message for message in messages if not _is_system_instruction(message)
    ]
    for message in non_system_messages:
        if "role" in message:
            normalized_role = normalize_message_role(message.get("role"))  # type: ignore
            content = message.get("content")  # type: ignore
            request_messages.append(
                {
                    "role": normalized_role,
                    "content": (
                        [{"type": "text", "text": content}]
                        if isinstance(content, str)
                        else content
                    ),
                }
            )
        else:
            if message.get("type") == "function_call":  # type: ignore
                request_messages.append(
                    {
                        "role": GEN_AI_ALLOWED_MESSAGE_ROLES.ASSISTANT,
                        "content": [message],
                    }
                )
            elif message.get("type") == "function_call_output":  # type: ignore
                request_messages.append(
                    {
                        "role": GEN_AI_ALLOWED_MESSAGE_ROLES.TOOL,
                        "content": [message],
                    }
                )

normalized_messages = normalize_message_roles(request_messages)
    client = sentry_sdk.get_client()
    scope = sentry_sdk.get_current_scope()
    messages_data = (
        normalized_messages
        if client.options.get("stream_gen_ai_spans", False)
        else truncate_and_annotate_messages(normalized_messages, span, scope)
    )
    if messages_data is not None:
        set_data_normalized(
            span,
            SPANDATA.GEN_AI_REQUEST_MESSAGES,
            messages_data,
            unpack=False,
        )

def _set_output_data(span: "sentry_sdk.tracing.Span", result: "Any") -> None:
    if not should_send_default_pii():
        return

output_messages: "dict[str, list[Any]]" = {
        "response": [],
        "tool": [],
    }

for output in result.output:
        if output.type == "function_call":
            output_messages["tool"].append(output.dict())
        elif output.type == "message":
            for output_message in output.content:
                try:
                    output_messages["response"].append(output_message.text)
                except AttributeError:
                    # Unknown output message type, just return the json
                    output_messages["response"].append(output_message.dict())

if len(output_messages["tool"]) > 0:
        span.set_data(
            SPANDATA.GEN_AI_RESPONSE_TOOL_CALLS, safe_serialize(output_messages["tool"])
        )

if len(output_messages["response"]) > 0:
        set_data_normalized(
            span, SPANDATA.GEN_AI_RESPONSE_TEXT, output_messages["response"]
        )

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Directory Contents

Dirs: 3 Ã— Files: 3

Sort:

Enable drag-to-move

Name	Size	Perms	Modified	Actions
patches DIR	-	drwxr-xr-x	2026-06-11 06:30:31	Edit Download recursive recursive
spans DIR	-	drwxr-xr-x	2026-06-11 06:30:31	Edit Download recursive recursive
__pycache__ DIR	-	drwxr-xr-x	2026-06-11 06:30:31	Edit Download recursive recursive
consts.py	38 B	lrw-r--r--	2026-06-11 06:30:31	Edit Download recursive recursive
utils.py	7.22 KB	lrw-r--r--	2026-06-11 06:30:31	Edit Download recursive recursive
__init__.py	10.82 KB	lrw-r--r--	2026-06-11 06:30:31	Edit Download recursive recursive