AWS Bedrock

AWS Bedrock API Integration

Integrates Amazon Bedrock using the AWS SDK for Python (Boto3). When loaded, this skill makes the model implement Bedrock API calls using the Converse API (recommended cross-model interface), InvokeModel (native payloads), Knowledge Bases, Agents, and Guardrails with proper IAM authentication and error handling.

When to Use

Use this skill when:

• Building applications that call foundation models through Amazon Bedrock
• Using the Converse API for a unified interface across Claude, Llama, Titan, Nova, Mistral, Cohere, and DeepSeek
• Integrating Bedrock Knowledge Bases for RAG with fully managed data ingestion
• Building Bedrock Agents with multi-step orchestration
• Implementing Guardrails for content filtering and responsible AI
• Working with Amazon Nova models (Micro, Lite, Pro) for cost-effective inference

When NOT to Use

• For calling Anthropic Claude directly (not via Bedrock), use coding-anthropic-api
• For OpenAI API calls, use coding-openai-api
• For general Boto3 patterns unrelated to Bedrock, refer to AWS SDK documentation

Core Workflow

1. Configure AWS Credentials — Set up AWS credentials via environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION), AWS CLI config, or IAM roles. Ensure the IAM user/role has bedrock:InvokeModel and bedrock:ListFoundationModels permissions. Checkpoint: Verify access by listing models with boto3.client("bedrock").list_foundation_models().

2. Select the API — Converse vs. InvokeModel — The Converse API (client.converse()) is the recommended unified interface. It normalizes requests and responses across all supported models. Use InvokeModel (client.invoke_model()) only when you need model-specific features not exposed by Converse. Checkpoint: Check if your model supports Converse by looking for "responseStreamingSupported" in the model summary.

3. Send a Message with Converse API — Create a bedrock-runtime client and call converse() with modelId, messages, and inferenceConfig. The response format is uniform across all models: extract text from response["output"]["message"]["content"][0]["text"]. Checkpoint: Test with Amazon Nova Micro (fastest, cheapest) to validate the pipeline before using more expensive models.

4. Implement Tool Use (Function Calling) — Define tools as a list of {"toolSpec": {"name": ..., "description": ..., "inputSchema": ...}} dicts and pass them to converse(). Handle toolUse blocks in the response, execute tools, and return results. Checkpoint: Verify stopReason is "tool_use" when the model requests tools.

5. Integrate Knowledge Bases — Use bedrock-agent-runtime client with retrieve() or retrieve_and_generate() to query knowledge bases. Configure the KB ID, query text, and number of results. Checkpoint: Verify that retrieve() returns retrievalResults with content and location fields.

Implementation Patterns

Pattern 1: Converse API (Recommended Cross-Model Interface)

from __future__ import annotations

import boto3
from botocore.exceptions import ClientError

# ❌ BAD — model-specific payload, no error handling, InvokeModel (non-portable)
client = boto3.client("bedrock-runtime")
body = json.dumps({
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 1024,
    "messages": [{"role": "user", "content": [{"type": "text", "text": "Hello"}]}],
})
response = client.invoke_model(modelId="anthropic.claude-3-sonnet-20240229-v1:0", body=body)
print(json.loads(response["body"].read())["content"][0]["text"])

# ✅ GOOD — Converse API, typed, cross-model, proper error handling
class BedrockClient:
    """AWS Bedrock client using the unified Converse API."""

    def __init__(self, region: str = "us-east-1") -> None:
        self.runtime = boto3.client("bedrock-runtime", region_name=region)
        self.control = boto3.client("bedrock", region_name=region)

    def list_available_models(self) -> list[dict]:
        """List foundation models accessible in this account/region."""
        response = self.control.list_foundation_models()
        return response.get("modelSummaries", [])

    def converse(
        self,
        prompt: str,
        model_id: str = "amazon.nova-micro-v1:0",
        max_tokens: int = 512,
        temperature: float = 0.5,
    ) -> str:
        """Send a prompt via the Converse API.

        Args:
            prompt: The user input text.
            model_id: Bedrock model ID (e.g., amazon.nova-micro-v1:0,
                     anthropic.claude-3-sonnet-20240229-v1:0,
                     meta.llama3-70b-instruct-v1:0).
            max_tokens: Maximum output tokens.
            temperature: Sampling temperature.

        Returns:
            The model's response text.

        Raises:
            RuntimeError: On API access or permission errors.
        """
        conversation = [{"role": "user", "content": [{"text": prompt}]}]

        try:
            response = self.runtime.converse(
                modelId=model_id,
                messages=conversation,
                inferenceConfig={
                    "maxTokens": max_tokens,
                    "temperature": temperature,
                },
            )
            output = response["output"]["message"]["content"]
            return "".join(block["text"] for block in output if "text" in block)
        except ClientError as e:
            error_code = e.response["Error"]["Code"]
            if error_code == "AccessDeniedException":
                raise PermissionError(
                    "Bedrock access denied. Check IAM permissions and model access."
                ) from e
            if error_code == "ThrottlingException":
                raise RuntimeError("Bedrock rate limit exceeded. Implement retry.") from e
            raise RuntimeError(f"Bedrock API error: {error_code}: {e}") from e

Pattern 2: Knowledge Base Retrieval (RAG)

from __future__ import annotations

import boto3
from botocore.exceptions import ClientError


class BedrockKnowledgeBase:
    """Query Bedrock Knowledge Bases for RAG."""

    def __init__(self, region: str = "us-east-1") -> None:
        self.agent_runtime = boto3.client("bedrock-agent-runtime", region_name=region)

    def retrieve(
        self,
        knowledge_base_id: str,
        query: str,
        top_k: int = 5,
    ) -> list[dict]:
        """Retrieve relevant documents from a Bedrock Knowledge Base.

        Args:
            knowledge_base_id: The KB UUID.
            query: Natural language query.
            top_k: Number of results to return.

        Returns:
            List of retrieval results with content and metadata.

        Raises:
            ValueError: If the KB ID is invalid or query fails.
        """
        try:
            response = self.agent_runtime.retrieve(
                knowledgeBaseId=knowledge_base_id,
                retrievalQuery={"text": query},
                retrievalConfiguration={
                    "vectorSearchConfiguration": {
                        "numberOfResults": top_k,
                    }
                },
            )
            return response.get("retrievalResults", [])
        except ClientError as e:
            raise ValueError(f"Knowledge base retrieval failed: {e}") from e

    def retrieve_and_generate(
        self,
        knowledge_base_id: str,
        query: str,
        model_id: str = "amazon.nova-micro-v1:0",
    ) -> str:
        """Retrieve context from KB and generate a response.

        Args:
            knowledge_base_id: The KB UUID.
            query: Natural language query.
            model_id: The Bedrock model to use for generation.

        Returns:
            Generated response grounded in KB context.
        """
        try:
            response = self.agent_runtime.retrieve_and_generate(
                knowledgeBaseId=knowledge_base_id,
                input={"text": query},
                generationConfiguration={
                    "generationModel": {"modelId": model_id},
                },
            )
            return response["output"]["text"]
        except ClientError as e:
            raise RuntimeError(f"Retrieve and generate failed: {e}") from e

Pattern 3: Tool Use with Converse API

from __future__ import annotations

import json
from typing import Any

import boto3


def converse_with_tools(
    prompt: str,
    model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0",
) -> str:
    """Use the Converse API with tool/function calling.

    Args:
        prompt: User input.
        model_id: Bedrock model ID that supports tool use.

    Returns:
        Final response after tool execution.
    """
    client = boto3.client("bedrock-runtime")

    tools: list[dict[str, Any]] = [
        {
            "toolSpec": {
                "name": "get_weather",
                "description": "Get current weather for a location",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {
                            "location": {
                                "type": "string",
                                "description": "City and state",
                            }
                        },
                        "required": ["location"],
                    }
                },
            }
        }
    ]

    messages: list[dict[str, Any]] = [
        {"role": "user", "content": [{"text": prompt}]}
    ]

    response = client.converse(
        modelId=model_id,
        messages=messages,
        toolConfig={"tools": tools},
    )

    stop_reason = response.get("stopReason")

    if stop_reason == "tool_use":
        # Extract tool calls
        content = response["output"]["message"]["content"]
        for block in content:
            if "toolUse" in block:
                tool = block["toolUse"]
                if tool["name"] == "get_weather":
                    args = tool["input"]
                    weather = {"temperature": 72, "condition": "sunny"}
                    messages.append({
                        "role": "user",
                        "content": [{
                            "toolResult": {
                                "toolUseId": tool["toolUseId"],
                                "content": [{"json": weather}],
                            }
                        }],
                    })

        # Get final response
        final = client.converse(
            modelId=model_id,
            messages=messages,
            toolConfig={"tools": tools},
        )
        text_blocks = final["output"]["message"]["content"]
        return "".join(b["text"] for b in text_blocks if "text" in b)

    text_blocks = response["output"]["message"]["content"]
    return "".join(b["text"] for b in text_blocks if "text" in b)

Constraints

MUST DO

• Use the Converse API (client.converse()) over invoke_model() for cross-model portability
• Use region_name parameter when creating Bedrock clients (default may not have Bedrock enabled)
• Check stopReason in Converse responses to determine if the model requested tool use
• Handle ClientError with specific error code checks (AccessDeniedException, ThrottlingException, ValidationException)
• Verify model access in the AWS Console before coding — models must be explicitly enabled per region

MUST NOT DO

• Hardcode AWS credentials in source files — use environment variables, AWS CLI config, or IAM roles
• Assume all models support the Converse API — check responseStreamingSupported in model summary
• Use InvokeModel with model-specific payloads when Converse is available
• Skip pagination when listing knowledge bases or models — use Boto3 paginators

Live References

Related Skills