document_upload_sample.py

"""Upload 10 PDF documents to a Strands agent using AnthropicBedrockModel.
Demonstrates that the Anthropic Messages API (via AnthropicBedrockModel) bypasses
the Bedrock Converse API's 5-document-per-request limit.
"""

from pathlib import Path

import strands
from anthropic_bedrock_model import AnthropicBedrockModel
from fpdf import FPDF
from strands.types.content import ContentBlock

NUM_DOCUMENTS = 10
DOCS_DIR = Path(__file__).parent / "sample_docs"


def generate_pdfs() -> None:
    """Generate 10 sample PDF files and write them to the sample_docs folder."""
    DOCS_DIR.mkdir(exist_ok=True)

    for i in range(1, NUM_DOCUMENTS + 1):
        title = f"Document {i}"
        body = (
            f"This is sample document number {i}. "
            f"It contains unique information: the secret code for document {i} is CODE-{i * 111}."
        )

        pdf = FPDF()
        pdf.add_page()
        pdf.set_font("Helvetica", "B", 16)
        pdf.cell(text=title)
        pdf.ln(12)
        pdf.set_font("Helvetica", size=12)
        pdf.multi_cell(w=0, text=body)

        path = DOCS_DIR / f"document_{i}.pdf"
        pdf.output(str(path))
        print(f"  Created {path.name}")


def build_content_blocks() -> list[ContentBlock]:
    """Read PDFs from the sample_docs folder and build ContentBlocks."""
    blocks: list[ContentBlock] = []

    for path in sorted(DOCS_DIR.glob("*.pdf")):
        pdf_bytes = path.read_bytes()
        blocks.append(
            ContentBlock(
                document={
                    "format": "pdf",
                    "name": path.stem,
                    "source": {"bytes": pdf_bytes},
                }
            )
        )

    blocks.append(
        ContentBlock(
            text=(
                f"I have uploaded {len(blocks)} PDF documents. "
                "Please confirm how many documents you received, list each document by name, "
                "and tell me the secret code found in each one."
            )
        )
    )

    return blocks


def main() -> None:
    print(f"Generating {NUM_DOCUMENTS} PDF documents in {DOCS_DIR}/...")
    generate_pdfs()

    model = AnthropicBedrockModel(
        aws_region="us-east-1",
        model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
        max_tokens=4096,
    )

    agent = strands.Agent(model=model)
    content_blocks = build_content_blocks()

    print(f"\nSending {NUM_DOCUMENTS} PDF documents to the agent...\n")
    response = agent(content_blocks)
    print(response)


if __name__ == "__main__":
    main()