Pydantic and dataclasses

Define application inputs and outputs around an async function that calls rboto. You choose the fields and validation rules. Client methods keep their generated signatures, and responses keep their native, read-only attributes.

Pydantic inputs and outputs

Install Pydantic 2 alongside the clients you use:

python -m pip install "pydantic>=2,<3"

rboto does not require Pydantic. The following example assumes that you have installed the S3 client as described in Getting started.

This tool accepts an object key and returns the object's size and content type:

from pydantic import BaseModel, ConfigDict, Field
from rboto import S3Client


class DocumentInput(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")

    key: str = Field(min_length=1, description="Object key within the document bucket")


class DocumentOutput(BaseModel):
    model_config = ConfigDict(from_attributes=True, strict=True)

    content_length: int = Field(ge=0)
    content_type: str | None = None


async def inspect_document(
    client: S3Client,
    data: DocumentInput,
    *,
    bucket: str,
) -> DocumentOutput:
    response = await client.head_object(bucket=bucket, key=data.key)
    return DocumentOutput.model_validate(response)


async def run_tool(client: S3Client, arguments_json: str, *, bucket: str) -> str:
    data = DocumentInput.model_validate_json(arguments_json)
    result = await inspect_document(client, data, bucket=bucket)
    return result.model_dump_json()


input_schema = DocumentInput.model_json_schema()
output_schema = DocumentOutput.model_json_schema(mode="serialization")

The application supplies the client and bucket. The agent supplies only the arguments described by DocumentInput. Pass the generated schema using your agent framework's tool registration format; supported JSON Schema features vary by framework and model.

run_tool() validates the JSON before making a request, awaits the operation, validates the response, and returns JSON. strict=True disables coercions such as converting a string to an integer. extra="forbid" rejects unexpected input fields. Generated rboto method signatures still check the explicit bucket and key arguments when you run mypy or Pyright.

from_attributes=True lets Pydantic read the selected attributes directly from the native response. You do not need to call to_dict() first. Validation creates a separate Pydantic model; it does not change the response. Only fields selected by your model are requested, but reading a selected collection still materializes that collection.

content_length is optional in the service response. This application's output requires a nonnegative integer, so an omitted size causes a Pydantic ValidationError. Use int | None = None if omission is acceptable. Defaults apply to missing attributes; they do not replace an explicit None.

Service failures keep their rboto exception types. Handle those separately from input or output validation errors. See Errors.

Nested responses

Configure attribute validation on each nested model:

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field
from rboto import S3Client


class Bucket(BaseModel):
    model_config = ConfigDict(from_attributes=True, strict=True)

    name: str
    creation_date: datetime | None = Field(default=None, strict=False)


class BucketPage(BaseModel):
    model_config = ConfigDict(from_attributes=True, strict=True)

    buckets: list[Bucket] | None = None
    continuation_token: str | None = None


async def list_bucket_page(client: S3Client) -> BucketPage:
    response = await client.list_buckets()
    return BucketPage.model_validate(response)

This reads native bucket objects inside the native response. A bucket without a name fails validation because this application's Bucket requires one. BucketPage describes one response page; pagination remains explicit.

rboto exposes timestamps as strings. The creation_date field explicitly enables conversion to datetime with Field(strict=False). Keep a str annotation if you want to retain the original representation.

Standard dataclasses

Dataclasses work without Pydantic. Construct them from the response attributes you want to retain:

from dataclasses import dataclass

from rboto import S3Client


@dataclass(frozen=True, slots=True)
class DocumentInput:
    key: str


@dataclass(frozen=True, slots=True)
class DocumentOutput:
    content_length: int
    content_type: str | None = None


async def inspect_document(
    client: S3Client,
    data: DocumentInput,
    *,
    bucket: str,
) -> DocumentOutput:
    response = await client.head_object(bucket=bucket, key=data.key)
    if response.content_length is None:
        raise ValueError("The response did not include an object size")
    return DocumentOutput(
        content_length=response.content_length,
        content_type=response.content_type,
    )

Standard dataclass annotations do not validate values at runtime. frozen=True prevents assigning to fields after construction.

If you need JSON validation and a schema for a dataclass, add Pydantic's TypeAdapter:

from dataclasses import dataclass
from typing import Annotated

from pydantic import Field, TypeAdapter


@dataclass(frozen=True, slots=True)
class ToolInput:
    key: Annotated[str, Field(strict=True, min_length=1)]


# Construct once and reuse the adapter across tool calls.
input_adapter = TypeAdapter(ToolInput)
input_schema = input_adapter.json_schema()


def parse_arguments(arguments_json: str) -> ToolInput:
    return input_adapter.validate_json(arguments_json)

The adapter returns a ToolInput instance. Validation runs when you call the adapter; calling the standard dataclass constructor directly bypasses it. For native responses, construct the dataclass explicitly as above, or validate a dictionary with an adapter. Attribute validation for BaseModel is a separate path from dataclass validation.

For a complete handler that validates an event and returns JSON-compatible data, see AWS Lambda.

Structured model responses

For an agent using Bedrock, extract the JSON text produced by the model and pass it to your application's BaseModel.model_validate_json() or TypeAdapter.validate_json(). The Bedrock response envelope has a different structure from that application payload.

Validate returned data even when you supplied a schema to the model. Choosing tools, handling validation failures, and retrying model generation belong in the application or agent framework.

Streaming responses

Select metadata fields when validating a streaming response. Leave the body or event receiver out of the Pydantic model, then consume it explicitly with await response.body.read() or async iteration.

Validating metadata does not read the body. Streams are consumable state and do not become JSON through model validation. See Streaming.