raindrop-bedrock 0.0.3

Raindrop observability integration for AWS Bedrock

raindrop-bedrock

Raindrop observability integration for AWS Bedrock (Python). Automatically captures converse() and invoke_model() calls by wrapping the boto3 bedrock-runtime client.

Installation

pip install raindrop-bedrock

For async support with aioboto3:

pip install raindrop-bedrock[async]

Quick Start

import boto3
from raindrop_bedrock import RaindropBedrock

rb = RaindropBedrock(api_key="your-write-key", user_id="user-123")

client = boto3.client("bedrock-runtime", region_name="us-east-1")
rb.wrap(client)

response = client.converse(
    modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=[{"role": "user", "content": [{"text": "Hello!"}]}],
)

rb.flush()

Debug Mode

Enable verbose logging with the debug flag:

rb = RaindropBedrock(api_key="your-write-key", user_id="user-123", debug=True)

Async Usage

import aioboto3
from raindrop_bedrock import RaindropBedrock

rb = RaindropBedrock(api_key="rk_...", user_id="user-123")

session = aioboto3.Session()
async with session.client("bedrock-runtime", region_name="us-east-1") as client:
    rb.async_wrap(client)
    response = await client.converse(
        modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
        messages=[{"role": "user", "content": [{"text": "Hello!"}]}],
    )

rb.flush()

identify()

Associate a user with optional traits:

rb.identify(user_id="user-123", traits={"plan": "pro", "company": "Acme"})

track_signal()

Track feedback, edits, or custom signals:

rb.track_signal(
    event_id="evt_abc123",
    name="thumbs_up",
    signal_type="feedback",
    sentiment="POSITIVE",
    comment="Great answer!",
)

flush() / shutdown()

Always call flush() before your process exits to ensure all telemetry is shipped:

rb.flush()      # flush pending data
rb.shutdown()   # flush + release resources

Legacy Factory Function

The create_raindrop_bedrock() factory function is still supported for backwards compatibility:

from raindrop_bedrock import create_raindrop_bedrock

raindrop = create_raindrop_bedrock(api_key="your-write-key", user_id="user-123")
client = boto3.client("bedrock-runtime", region_name="us-east-1")
raindrop.wrap(client)

What Gets Captured

Method	Captured Data
converse()	Input messages, output text, model ID, token usage (inputTokens/outputTokens), stop reason (stopReason), cached tokens (cacheReadInputTokenCount, cacheWriteInputTokenCount), conversation ID
invoke_model()	Raw request/response bodies, model ID, token usage (Claude, Titan, and Llama formats), stop reason (Claude: stop_reason, Llama: stop_reason), cached tokens (Claude: cache_read_input_tokens)
Errors	Error type and message are captured in event properties, then the exception is re-raised

Captured Properties

Property Key	Source	Description
ai.usage.prompt_tokens	Both APIs	Input/prompt token count
ai.usage.completion_tokens	Both APIs	Output/completion token count
ai.usage.cached_tokens	Converse: cacheReadInputTokenCount; Claude InvokeModel: cache_read_input_tokens	Tokens read from cache
ai.usage.cache_write_tokens	Converse: cacheWriteInputTokenCount	Tokens written to cache
bedrock.finish_reason	Converse: stopReason; InvokeModel: varies by model	Why the model stopped generating

API Reference

RaindropBedrock(api_key=None, user_id=None, convo_id=None, tracing_enabled=True, bypass_otel_for_tools=True, debug=False)

Parameter	Type	Default	Description
api_key	str | None	None	Raindrop API key. Warns if not provided.
user_id	str | None	None	Default user ID for events (falls back to "unknown")
convo_id	str | None	None	Group events into a conversation
tracing_enabled	bool	True	Enable Raindrop tracing
bypass_otel_for_tools	bool	True	Bypass OpenTelemetry for tool-level instrumentation
debug	bool	False	Enable verbose DEBUG-level logging

Methods

Method	Description
wrap(client)	Instrument a sync boto3 bedrock-runtime client
async_wrap(client)	Instrument an async aioboto3 bedrock-runtime client
identify(user_id, traits=None)	Identify a user with optional traits
track_signal(event_id, name, ...)	Track a signal event
flush()	Flush pending events
shutdown()	Flush and shut down

Testing

cd packages/bedrock-python
pip install -e ".[async]"
pip install pytest
python -m pytest tests/ -v

Known Limitations

• InvokeModel body replacement: After consuming the response body stream, it's replaced with a BytesIO object. Callers using StreamingBody.read() will get the same bytes, but the original StreamingBody API is not preserved.
• Async support requires the [async] extra (aioboto3>=12.0.0).

Full Documentation

docs.raindrop.ai/integrations/bedrock

License

MIT