aiobedrock 0.3.1

AWS boto3 bedrock client in async

aiobedrock

An asynchronous Python client for AWS Bedrock, providing non-blocking access to Amazon's foundation model service.

Features

• Fully Asynchronous: Non-blocking API calls using aiohttp
• Low Overhead: Minimal dependencies with efficient implementation
• Streaming Support: Stream responses for real-time AI model interactions
• Guardrail Integration: Support for AWS Bedrock Guardrails
• AWS SigV4 Auth: Proper AWS authentication for secure API calls
• Error Handling: Comprehensive error handling with descriptive exceptions

Installation

pip install aiobedrock

Requirements

• Python 3.9 or later (tested through Python 3.14)
• AWS credentials configured in your environment
• boto3 1.35.84 or newer (installed automatically via dependencies)

Quick Start

Basic Model Invocation

import json
import asyncio
from aiobedrock import Client

async def main():
    async with Client(region_name="YOUR_AWS_REGION") as client:
        body = {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4096,
            "temperature": 0.7,
            "top_p": 0.9,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "What can you do?"},
                    ],
                }
            ],
        }

        response = await client.invoke_model(
            body=json.dumps(body),
            modelId="anthropic.claude-3-haiku-20240307-v1:0",
            accept="application/json",
            contentType="application/json",
        )

        print(json.loads(response.decode("utf-8")))

if __name__ == "__main__":
    asyncio.run(main())

Streaming Response

import json
import asyncio
from aiobedrock import Client

async def main():
    async with Client(region_name="YOUR_AWS_REGION") as client:
        body = {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4096,
            "temperature": 0.7,
            "top_p": 0.9,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "What can you do?"},
                    ],
                }
            ],
        }

        async for chunk in client.invoke_model_with_response_stream(
            body=json.dumps(body),
            modelId="anthropic.claude-3-haiku-20240307-v1:0",
            accept="application/json",
            contentType="application/json",
        ):
            print(json.loads(chunk.decode("utf-8")))

if __name__ == "__main__":
    asyncio.run(main())

Using Guardrails

import json
import asyncio
from aiobedrock import Client

async def main():
    async with Client(region_name="YOUR_AWS_REGION") as client:
        body = {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4096,
            "temperature": 0.7,
            "top_p": 0.9,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "What can you do?"},
                    ],
                }
            ],
        }

        response = await client.invoke_model(
            body=json.dumps(body),
            modelId="anthropic.claude-3-haiku-20240307-v1:0",
            accept="application/json",
            contentType="application/json",
            guardrailIdentifier="arn:aws:bedrock:YOUR_REGION:YOUR_ACCOUNT_ID:guardrail/YOUR_GUARDRAIL_ID",
            guardrailVersion="LATEST",
        )

        print(json.loads(response.decode("utf-8")))

if __name__ == "__main__":
    asyncio.run(main())

API Reference

Client

Client(region_name: str)

Creates a new Bedrock client instance.

The underlying aiohttp.ClientSession is created lazily when first used. You can interact with the client by using async with or by awaiting individual methods directly; both patterns will create a shared session automatically.

• region_name: AWS region where Bedrock is available (e.g., "us-east-1", "us-west-2", "ap-southeast-1")

Methods

invoke_model

async invoke_model(body: str, modelId: str, **kwargs) -> bytes

Invokes a Bedrock model and returns the complete response.

• body: JSON string with model parameters and prompt
• modelId: Bedrock model identifier
• kwargs: Optional parameters
  • accept: Accept header (default: "application/json")
  • contentType: Content-Type header (default: "application/json")
  • trace: Tracing level: "ENABLED", "ENABLED_FULL" or "DISABLED" (default: "DISABLED")
  • guardrailIdentifier: ARN of the guardrail to use
  • guardrailVersion: Version of the guardrail (e.g., "1" or "LATEST")
  • performanceConfigLatency: Performance configuration for latency. Valid values are "standard" or "optimized".

invoke_model_with_response_stream

async invoke_model_with_response_stream(body: str, modelId: str, **kwargs) -> AsyncGenerator[Union[Dict[str, Any], bytes], None]

Invokes a Bedrock model and returns an asynchronous generator. The generator yields either parsed JSON objects or raw byte chunks depending on the payload.

• Parameters are the same as invoke_model
• Streaming error events from Bedrock raise aiobedrock.main.BedrockStreamError and surface the error payload in the exception message so you can respond or retry appropriately.

invoke_many

async invoke_many(requests: Iterable[Mapping[str, Any]], *, concurrency: Optional[int] = None, return_exceptions: bool = False) -> Sequence[Union[bytes, Exception]]

Runs multiple invocations concurrently while preserving the order of results. Each entry in requests must include body (JSON string) and modelId; any additional key/value pairs are forwarded to invoke_model.

• concurrency: Optional per-call limit that overrides the client's global max_concurrency.
• return_exceptions: Mirrors asyncio.gather; when True, exceptions are returned alongside successful responses instead of aborting the batch.

See example/invoke_many.py for a complete usage example.

close

async close()

Closes the aiohttp session.

Supported Models

aiobedrock supports all models available on AWS Bedrock.

Ensure you have appropriate permissions to access these models in your AWS account.

Error Handling

The client provides detailed error messages for common Bedrock API errors:

• 403: AccessDeniedException
• 500: InternalServerException
• 424: ModelErrorException
• 408: ModelTimeoutException
• 429: ThrottlingException

In addition, when the streaming API surfaces an error event the library raises BedrockStreamError with the exception type that Bedrock reported (for example ModelStreamError) and the payload returned by the service.

For more error details, refer to the AWS Bedrock API documentation.

License

MIT License - See LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.