anth-reference-architecture

name	anth-reference-architecture
description	Implement Claude API reference architectures for common use cases. Use when designing a Claude-powered application, choosing between direct API vs queue-based, or planning a multi-model architecture. Trigger with phrases like "anthropic architecture", "claude system design", "anthropic reference architecture", "design claude integration".
allowed-tools	Read, Write, Edit, Grep
version	1.0.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","ai","anthropic"]
compatible-with	claude-code

Anthropic Reference Architecture

Overview

Three validated architecture patterns for Claude API integrations: synchronous API gateway, async queue-based processing, and multi-model routing.

Architecture 1: Sync API Gateway (Simple)

User → API Gateway → Claude Service → Messages API
                                     ↓
                                   Response → User

# Best for: chatbots, interactive tools, low-volume (<100 RPM)
from fastapi import FastAPI
import anthropic

app = FastAPI()
client = anthropic.Anthropic(max_retries=3, timeout=60.0)

@app.post("/chat")
async def chat(prompt: str):
    msg = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return {"text": msg.content[0].text, "tokens": msg.usage.output_tokens}

Architecture 2: Async Queue-Based (Scalable)

User → API → Queue (Redis/SQS) → Worker Pool → Messages API
  ↑                                                ↓
  └──────────── Status/Result ←── Result Store ←───┘

# Best for: batch processing, high-volume, background tasks
from redis import Redis
from rq import Queue
import anthropic

redis = Redis()
task_queue = Queue("claude-tasks", connection=redis)
result_store = Redis(db=1)

def process_task(task_id: , prompt: , model: ):
    client = anthropic.Anthropic()
    msg = client.messages.create(
        model=model,
        max_tokens=,
        messages=[{: , : prompt}]
    )
    result_store.setex(, , msg.content[].text)


 uuid
task_id = (uuid.uuid4())
task_queue.enqueue(process_task, task_id, prompt, )