bedrock-converse-client

Files (1)

• SKILL.md

SKILL.md (first lines)

<!-- Vendored from: platform-catalyst/skills/bedrock-converse-client/SKILL.md (BittahCriminal/platform-catalyst, BSD-3-Clause). Adapted for Catalyst: PLAN.md/CLAUDE.md/DECISIONS.md scrubbed; ADR-008->ADR-001, ADR-009->ADR-002. -->

# Bedrock Converse client

## Role

You guide implementation of Bedrock Converse API calls in Catalyst services, ensuring correct message structure, model selection, and error handling per `AGENTS.md` Bedrock-specific guidance.

## Instructions

### 1. Async client setup

```python
import aioboto3
from types import TracebackType

class BedrockClient:
    """Async Bedrock runtime client with connection reuse."""

    def __init__(self, region: str = "us-east-1"):
        self._session = aioboto3.Session()
        self._region = region
        self._client = None

    async def __aenter__(self):
        self._client = await self._session.client(
            "bedrock-runtime", region_name=self._region
        ).__aenter__()
        return self

    async def __aexit__(self, *exc):
        if self._client:
            await self._client.__aexit__(*exc)
```

### 2. Model IDs (runtime configuration)

```python
from dataclasses import dataclass
import os

@dataclass(frozen=True)
class BedrockModels:
    reviewer: str
    summarizer: str

def load_models() -> BedrockModels:
    return BedrockModels(
        reviewer=os.environ["CATALYST_BEDROCK_MODEL_REVIEWER"],
        summarizer=os.environ["CATALYST_BEDROCK_MODEL_SUMMARIZER"],
    )
```

### 3. Converse API call

```python
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from botocore.exceptions import ClientError

@retry(
    stop=stop_after_attempt(2),
    wait=wait_exponential(multiplier=1, min=1, max=5),
    retry=retry_if_exception_type(ClientError),
)
async def converse(
    self,
    model_id: str,
    system_prompt: str,
    user_message: str,
    max_tokens: int = 4096,
    temperature: float = 0.0,
) -> dict:
    """Call Bedrock Converse API with retry on transient errors."""
    response = await self._client.converse(
        modelId=model_id,
        messages=[
            {
                "role": "user",
                "content": [{"text": user_message}],
            }
        ],
        system=[{"text": system_prompt}],
        inferenceConfig={
            "maxTokens": max_tokens,
            "temperature": temperature,
        },
    )
    return response
```

### 4. Response extraction

```python
def extract_text(response: dict) -> str:
    """Extract text content from Converse API response."""
    output = response.get("output", {})
    message = output.get("message", {})
    content = message.get("content", [])
    return "".join(block.get("text", "") for block in content)

def extract_usage(response: dict) -> dict:
    """Extract token usage from Converse API response."""
    usage = response.get("usage", {})
    return {
        "input_tokens": usage.get("inputTokens", 0),
        "output_tokens": usage.get("outputTokens", 0),
        "total_tokens": usage.get("inputTokens", 0) + usage.get("outputTokens", 0),
    }
```

### 5. Streaming (ConverseStream)

```python
async def converse_stream(
    self,
    model_id: str,
    system_prompt: str,
    user_message: str,
    max_tokens: int = 4096,
) -> AsyncIterator[str]:

From Cloud-Byte-Consulting/Catalyst, BSD-3-Clause.

More from Cloud-Byte-Consulting/Catalyst