Metadata-Version: 2.4
Name: bedrock-kit
Version: 0.1.0
Summary: Small, opinionated AWS Bedrock client wrapper: adaptive throttle, cache-aware cost tracking, and structured-output parse-and-repair. Single-cloud, single-purpose.
Project-URL: Homepage, https://github.com/MukundaKatta/bedrock-kit
Project-URL: Issues, https://github.com/MukundaKatta/bedrock-kit/issues
Project-URL: Source, https://github.com/MukundaKatta/bedrock-kit
Author-email: Mukunda Rao Katta <mukunda.vjcs6@gmail.com>
License: MIT
License-File: LICENSE
Keywords: anthropic,aws,bedrock,claude,cost-tracking,llm,mlops,rag,retry,structured-output
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: pydantic>=2.0
Provides-Extra: boto
Requires-Dist: boto3>=1.28; extra == 'boto'
Provides-Extra: dev
Requires-Dist: pandas>=2.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Provides-Extra: pandas
Requires-Dist: pandas>=2.0; extra == 'pandas'
Description-Content-Type: text/markdown

# bedrock-kit

[![CI](https://github.com/MukundaKatta/bedrock-kit/actions/workflows/ci.yml/badge.svg)](https://github.com/MukundaKatta/bedrock-kit/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/bedrock-kit.svg)](https://pypi.org/project/bedrock-kit/)
[![Python](https://img.shields.io/pypi/pyversions/bedrock-kit.svg)](https://pypi.org/project/bedrock-kit/)
[![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)

**Small, opinionated AWS Bedrock client wrapper.**

Every Bedrock production team rebuilds the same three things: adaptive
throttle for `ThrottlingException`, per-call cost tracking that handles
cache-read tokens, and structured-output parsing-with-repair. `bedrock-kit`
ships those, and nothing else. Single-cloud, single-purpose. No proxy
server. Wraps the official boto3 client; you can inject a fake for testing.

## Install

```bash
pip install "bedrock-kit[boto]"
# optional
pip install "bedrock-kit[boto,pandas]"  # adds CostLedger.to_pandas()
```

## Quickstart

```python
from bedrock_kit import BedrockClient, AdaptiveThrottle, CostLedger

ledger = CostLedger()
client = BedrockClient(
    region="us-east-1",
    retry=AdaptiveThrottle(max_attempts=8, base_delay=1.0, max_delay=30.0),
    cost_ledger=ledger,
)

resp = client.invoke(
    model_id="anthropic.claude-sonnet-4-5",
    messages=[{"role": "user", "content": "Hello"}],
    system="You are concise.",
    max_tokens=512,
)

resp.text                       # the model's reply
resp.usage.input_tokens         # 12
resp.usage.cache_read_tokens    # 0
resp.cost_usd                   # 0.000176
resp.cache_hit                  # False
ledger.total_usd                # accumulates across all calls
ledger.by_model                 # {"anthropic.claude-sonnet-4-5": 0.000176}
```

## Structured output with repair

```python
from pydantic import BaseModel
from bedrock_kit import JsonSchema

class Sentiment(BaseModel):
    label: str
    confidence: float

resp = client.invoke(
    model_id="anthropic.claude-sonnet-4-5",
    messages=[{"role": "user", "content": "Classify: 'this is great!'"}],
    response_schema=JsonSchema(Sentiment, max_repair_attempts=2),
)
resp.parsed                  # Sentiment(label="positive", confidence=0.95)
```

If the model returns invalid JSON, `bedrock-kit` first does a light local
repair pass (strip markdown fences, trailing commas, surrounding prose).
If that still fails, it asks the model to fix its own output, up to
`max_repair_attempts` times.

## Adaptive throttle

```python
throttle = AdaptiveThrottle(
    max_attempts=8,    # total attempts (incl. first)
    base_delay=1.0,    # seconds
    max_delay=30.0,
    jitter=True,       # full-jitter (uniform(0, capped_delay))
)
```

Retries on Bedrock throttle codes: `ThrottlingException`,
`TooManyRequestsException`, `ServiceUnavailableException`,
`ProvisionedThroughputExceededException`, `ModelTimeoutException`. Does
**not** retry validation, auth, or model-not-found errors - those are
your bugs to fix, not transient.

## Cost tracking

`CostLedger` ships pricing for popular Anthropic Bedrock models. Override
or extend with `pricing=`:

```python
from bedrock_kit import CostLedger, Pricing

ledger = CostLedger(
    pricing={
        "amazon.nova-pro-v1:0": Pricing(
            input=0.8, output=3.2, cache_read=0.2, cache_write=1.0
        ),
    },
    strict=True,  # raise PricingNotFoundError on unknown models
)
```

Default pricing is best-effort and dated; verify against
[aws.amazon.com/bedrock/pricing](https://aws.amazon.com/bedrock/pricing)
before using these numbers for billing.

## Why not LiteLLM?

LiteLLM is great if you need cross-provider routing. `bedrock-kit` is for
the case where you've already decided on Bedrock, you don't want a 46k-LOC
multi-provider abstraction, and you want a small surface a security team
can audit. We don't proxy, don't include a server, don't ship a UI. We're
< 1000 LOC of Python.

## What it explicitly does NOT do

- No multi-provider routing
- No proxy server, no UI
- No prompt management
- No agent loop
- No image generation
- No SageMaker, Bedrock Agents, or Knowledge Bases SDK wrapping
- No streaming or cancellation yet (planned for v0.2)
- No OpenTelemetry emission yet (planned for v0.2)

## Testing without AWS

The default `BedrockClient` makes a real boto3 client. For tests, inject a
fake that quacks like one:

```python
from bedrock_kit import BedrockClient, AdaptiveThrottle

class FakeClient:
    def converse(self, **kwargs):
        return {"output": {"message": {"content": [{"text": "stub"}]}},
                "stopReason": "end_turn",
                "usage": {"inputTokens": 1, "outputTokens": 1}}

client = BedrockClient(client=FakeClient(), retry=AdaptiveThrottle(sleep_fn=lambda _: None))
```

## Status

v0.1 - alpha. Public API may change before v1.0. Issues and PRs welcome.
