OpenTelemetry Bedrock Instrumentation

from opentelemetry.instrumentation.bedrock import BedrockInstrumentor

from opentelemetry.instrumentation.bedrock.config import Config
from opentelemetry.instrumentation.bedrock.event_models import MessageEvent, ChoiceEvent, Roles
from opentelemetry.instrumentation.bedrock.utils import should_send_prompts, should_emit_events
from opentelemetry.instrumentation.bedrock.guardrail import is_guardrail_activated, GuardrailAttributes, Type
from opentelemetry.instrumentation.bedrock.prompt_caching import CachingHeaders, CacheSpanAttrs

import boto3
from opentelemetry.instrumentation.bedrock import BedrockInstrumentor

# Basic instrumentation - automatically instruments all bedrock-runtime clients
BedrockInstrumentor().instrument()

# Create a Bedrock client (will be automatically instrumented)
bedrock_client = boto3.client(
    service_name='bedrock-runtime',
    region_name='us-east-1'
)

# Use the client normally - instrumentation happens automatically
response = bedrock_client.invoke_model(
    modelId='anthropic.claude-3-sonnet-20240229-v1:0',
    body=json.dumps({
        "messages": [{"role": "user", "content": "Hello, world!"}],
        "max_tokens": 100,
        "anthropic_version": "bedrock-2023-05-31"
    })
)

# Advanced configuration with custom options
BedrockInstrumentor(
    enrich_token_usage=True,
    use_legacy_attributes=False,
    exception_logger=my_exception_handler
).instrument()

class BedrockInstrumentor(BaseInstrumentor):
    def __init__(
        self,
        enrich_token_usage: bool = False,
        exception_logger = None,
        use_legacy_attributes: bool = True
    ): ...
    
    def instrument(self, **kwargs): ...
    def uninstrument(self, **kwargs): ...

@dataclass
class MessageEvent:
    content: Any
    role: str = "user"
    tool_calls: List[ToolCall] | None = None

@dataclass  
class ChoiceEvent:
    index: int
    message: CompletionMessage
    finish_reason: str = "unknown"
    tool_calls: List[ToolCall] | None = None

def emit_message_events(event_logger, kwargs): ...
def emit_choice_events(event_logger, response): ...

class MetricParams:
    def __init__(
        self,
        token_histogram: Histogram,
        choice_counter: Counter,
        duration_histogram: Histogram,
        exception_counter: Counter,
        guardrail_activation: Counter,
        guardrail_latency_histogram: Histogram,
        # ... additional guardrail and caching metrics
    ): ...

def is_metrics_enabled() -> bool: ...

class GuardrailMeters:
    LLM_BEDROCK_GUARDRAIL_ACTIVATION = "gen_ai.bedrock.guardrail.activation"
    LLM_BEDROCK_GUARDRAIL_LATENCY = "gen_ai.bedrock.guardrail.latency"
    # ... additional constants

def dont_throw(func): ...
def should_send_prompts() -> bool: ...  
def should_emit_events() -> bool: ...

class StreamingWrapper(ObjectProxy):
    def __init__(self, response, stream_done_callback): ...
    def __iter__(self): ...

class ReusableStreamingBody(StreamingBody):
    def __init__(self, raw_stream, content_length): ...
    def read(self, amt=None): ...

class Config:
    enrich_token_usage: bool
    exception_logger: Any
    use_legacy_attributes: bool

class GuardrailAttributes:
    """Constants for guardrail span attributes in OpenTelemetry instrumentation."""
    GUARDRAIL = "gen_ai.guardrail"
    TYPE = "gen_ai.guardrail.type"
    PII = "gen_ai.guardrail.pii"
    PATTERN = "gen_ai.guardrail.pattern"
    TOPIC = "gen_ai.guardrail.topic"
    CONTENT = "gen_ai.guardrail.content"
    CONFIDENCE = "gen_ai.guardrail.confidence"
    MATCH = "gen_ai.guardrail.match"

from enum import Enum

class Type(Enum):
    """Enum for guardrail assessment types."""
    INPUT = "input"
    OUTPUT = "output"

class CachingHeaders:
    """HTTP headers for Bedrock prompt caching functionality."""
    READ = "x-amzn-bedrock-cache-read-input-token-count"
    WRITE = "x-amzn-bedrock-cache-write-input-token-count"

class CacheSpanAttrs:
    """Span attributes for prompt caching operations."""
    TYPE = "gen_ai.cache.type"
    CACHED = "gen_ai.prompt_caching"

from enum import Enum

class Roles(Enum):
    """Valid roles for message events in conversation."""
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"
    TOOL = "tool"