Anthropic Examples

Code examples for common Claude integration patterns.

---

Image Captioning

Basic Caption Request

import anthropic
import base64
from pathlib import Path

def caption_image(image_path: str, api_key: str) -> dict:
    client = anthropic.Client(api_key=api_key)

    # Prepare image
    image_data = Path(image_path).read_bytes()
    base64_image = base64.standard_b64encode(image_data).decode("utf-8")

    # Detect MIME type (simplified)
    mime_type = "image/jpeg" if image_path.endswith(".jpg") else "image/png"

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1500,
        temperature=0.5,
        system="You are a fashion analyst. Describe garments technically.",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": mime_type,
                        "data": base64_image,
                    },
                },
                {
                    "type": "text",
                    "text": 'Describe this garment. Return JSON: {"technical_description": "...", "short_description": "..."}'
                }
            ],
        }],
    )

    return json.loads(response.content[0].text)

Caption with URL

def caption_from_url(image_url: str, api_key: str) -> dict:
    client = anthropic.Client(api_key=api_key)

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1500,
        temperature=0.5,
        system="You are a fashion analyst. Describe garments technically.",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "url",
                        "url": image_url,
                    },
                },
                {
                    "type": "text",
                    "text": 'Describe this garment. Return JSON: {"technical_description": "...", "short_description": "..."}'
                }
            ],
        }],
    )

    return json.loads(response.content[0].text)

---

Outfit Generation

Generate Outfit Recommendations

import json

def generate_outfit(
    core_product: dict,
    catalog: list[dict],
    api_key: str
) -> dict:
    client = anthropic.Client(api_key=api_key)

    # Format catalog as TSV for efficient token usage
    catalog_tsv = "id\tname\tcategory\tcolor\n"
    for item in catalog:
        catalog_tsv += f"{item['id']}\t{item['name']}\t{item['category']}\t{item['color']}\n"

    system_prompt = """
    You are a professional fashion stylist.
    Create outfit recommendations using ONLY products from the provided catalog.
    Each outfit must include the core product.
    Return valid JSON with product IDs from the catalog.
    """

    user_prompt = f"""
    Core product: {json.dumps(core_product)}

    Available catalog:
    {catalog_tsv}

    Create 3 complete outfit looks. Return:
    {{
        "looks": [
            {{
                "name": "Look name",
                "items": [{{"id": "uuid", "role": "category"}}]
            }}
        ]
    }}
    """

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4000,
        temperature=0.7,
        system=system_prompt,
        messages=[{"role": "user", "content": user_prompt}],
    )

    return json.loads(response.content[0].text)

---

Vision Analysis

Multi-Image Comparison

def compare_images(
    image_paths: list[str],
    prompt: str,
    api_key: str
) -> str:
    client = anthropic.Client(api_key=api_key)

    content = []

    # Add all images
    for path in image_paths:
        image_data = Path(path).read_bytes()
        base64_image = base64.standard_b64encode(image_data).decode("utf-8")
        content.append({
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": "image/jpeg",
                "data": base64_image,
            },
        })

    # Add analysis prompt
    content.append({
        "type": "text",
        "text": prompt
    })

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        temperature=0.3,
        messages=[{"role": "user", "content": content}],
    )

    return response.content[0].text

Style Consistency Check

def check_style_consistency(
    reference_image: str,
    generated_images: list[str],
    api_key: str
) -> dict:
    prompt = """
    Compare the reference image (first) with the generated images.

    For each generated image, evaluate:
    1. Style consistency (1-10)
    2. Color palette match (1-10)
    3. Overall quality (1-10)
    4. Issues found

    Return JSON:
    {
        "evaluations": [
            {"image_index": 1, "style": 8, "color": 7, "quality": 9, "issues": []}
        ]
    }
    """

    all_images = [reference_image] + generated_images
    return json.loads(compare_images(all_images, prompt, api_key))

---

Error Handling

Robust JSON Parsing

import json
import re

def parse_claude_json(response_text: str) -> dict:
    """Parse JSON from Claude response, handling various formats."""
    text = response_text.strip()

    # Try direct parse first
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    # Handle markdown code blocks
    patterns = [
        r'```json\s*(.*?)\s*```',
        r'```\s*(.*?)\s*```',
        r'\{.*\}',
    ]

    for pattern in patterns:
        match = re.search(pattern, text, re.DOTALL)
        if match:
            try:
                return json.loads(match.group(1) if '```' in pattern else match.group(0))
            except json.JSONDecodeError:
                continue

    raise ValueError(f"Could not parse JSON from response: {text[:200]}")

Retry with Backoff

import time
from anthropic import RateLimitError, APIError

def call_with_retry(func, max_retries: int = 3):
    """Call function with exponential backoff on rate limits."""
    for attempt in range(max_retries):
        try:
            return func()
        except RateLimitError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"Rate limited, waiting {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise
        except APIError as e:
            if "overloaded" in str(e).lower() and attempt < max_retries - 1:
                time.sleep(5)
            else:
                raise

---

Integration with AI Core

Using the Provider Abstraction

from ai_core.client import AICore
from ai_core.providers import AnthropicProvider
from ai_core.models import ProviderRequest, ModelConfig

# Initialize
provider = AnthropicProvider(api_key=settings.ANTHROPIC_API_KEY)
client = AICore(provider=provider)

# Create request
request = ProviderRequest(
    prompt="Describe this fashion photograph",
    system_prompt="You are a fashion analyst",
    images=["/path/to/image.jpg"],
    temperature=0.5,
    model_config=ModelConfig(
        provider="anthropic",
        model_name="claude-sonnet-4-20250514",
    ),
)

# Execute
response = await client.generate(request)
print(response.text)

---

Complete Workflow Example

Product Captioning Pipeline

import asyncio
from pathlib import Path

async def caption_products(
    product_images: list[dict],
    api_key: str,
    concurrency: int = 5
) -> list[dict]:
    """Caption multiple products with controlled concurrency."""

    semaphore = asyncio.Semaphore(concurrency)
    results = []

    async def process_one(product: dict):
        async with semaphore:
            caption = await asyncio.to_thread(
                caption_image,
                product["image_path"],
                api_key
            )
            return {
                "product_id": product["id"],
                "caption": caption
            }

    tasks = [process_one(p) for p in product_images]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter out errors
    successful = [r for r in results if not isinstance(r, Exception)]
    errors = [r for r in results if isinstance(r, Exception)]

    if errors:
        print(f"Warning: {len(errors)} captions failed")

    return successful

# Usage
products = [
    {"id": "prod-1", "image_path": "/images/dress.jpg"},
    {"id": "prod-2", "image_path": "/images/jacket.jpg"},
]

captions = asyncio.run(caption_products(products, api_key))