DEV Community: Peyton Green

Python Automation Cookbook, Part 2: Using AI to make decisions in your automation pipelines

Peyton Green — Fri, 10 Jul 2026 19:18:48 +0000

Part 1 covered the 25 scripts I reach for every week — reliable, production-ready automation building blocks. This part covers the part that takes longer to figure out: how to chain them together when the control flow involves a decision.

Pure script-chaining is easy. Run A, pipe output to B, done. But real automation pipelines hit decision points: should this file be processed or skipped? Is this API response an error, a retry, or a success? Does this content need review or can it proceed?

Historically you handled this with if/elif trees. This works fine when the decision logic is clear and stable. It falls apart when the decision requires judgment — parsing ambiguous output, classifying content that doesn't fit neat categories, handling the long tail of edge cases that if statements don't cover.

The pattern I've landed on: use the scripts in the Cookbook for the deterministic parts of the pipeline (HTTP calls, file I/O, scheduling, retries), and use a small AI prompt for the judgment calls. The two fit together cleanly because neither is trying to do the other's job.

The shape of an AI-in-the-loop pipeline

Here's a three-script pipeline before and after adding an AI decision layer.

Before — pure script chaining:

# Download files
python http_client.py --url $API_ENDPOINT --output raw/

# Process each file
for f in raw/*.json; do
    python csv_converter.py --input $f --output processed/
done

# Archive
python file_archiver.py --source processed/ --destination archive/

This works if every file from the API is well-formed JSON that maps cleanly to CSV. In practice: some files have malformed encoding, some have unexpected schema changes, some are empty, some are error responses that look like successful ones. The script chain fails silently on these, or fails loudly and stops the whole pipeline.

After — with an AI triage step:

# Download files
python http_client.py --url $API_ENDPOINT --output raw/

# AI triage: classify each file before processing
python prompt_runner.py \
  --prompt "classify_file_quality" \
  --input-dir raw/ \
  --output triage_report.json

# Route based on triage result
python pipeline_router.py \
  --triage triage_report.json \
  --good-dest processing/ \
  --bad-dest review/ \
  --skip-dest archive/

# Process clean files only
for f in processing/*.json; do
    python csv_converter.py --input $f --output processed/
done

The AI triage step doesn't replace the Cookbook scripts — it routes between them. That's the pattern.

What the AI step actually does

The triage prompt is small. Here's the actual prompt I use for file quality classification:

CLASSIFY_FILE_QUALITY = """You are a data quality classifier. Given a JSON file's contents, classify it as one of:
- PROCESS: well-formed, expected schema, ready for conversion
- REVIEW: unusual structure, possible schema change, needs human look
- SKIP: empty, error response, or duplicate of a file already processed

Respond with exactly one word: PROCESS, REVIEW, or SKIP.
Do not explain. Do not add caveats.

File contents:
{file_contents}
"""

Three outputs. No explanation. The AI is not being asked to think — it's being asked to classify. This is the right scope for an in-pipeline AI call.

The key constraints:

Single-word output — downstream code parses this; ambiguity breaks the pipeline
No reasoning trail — you don't need it and it slows the call
Enumerated choices — if you give the model open-ended output, you'll get open-ended answers

The pattern generalizes. You can use the same shape for:

Classifying log entries (ERROR / WARNING / INFO / NOISE)
Scoring extracted data quality (HIGH / MEDIUM / LOW / REJECT)
Routing customer messages to the right queue (BILLING / SUPPORT / FEEDBACK / IGNORE)
Deciding whether a scraped page contains the content type you're looking for

The prompt_runner.py script

The Cookbook includes prompt_runner.py — a lightweight wrapper that handles the boilerplate for in-pipeline AI calls: API client initialization, token budgeting, error handling, output parsing, and retry logic.

Here's how it works:

# Usage:
python prompt_runner.py \
  --prompt classify_file_quality \
  --input-file raw/data_2026_03_23.json \
  --output-file triage/data_2026_03_23.triage

# Batch mode (processes all files in a directory):
python prompt_runner.py \
  --prompt classify_file_quality \
  --input-dir raw/ \
  --output-dir triage/ \
  --workers 4

The --prompt flag takes a prompt name from prompts/ directory (where you store your project's prompt templates). It loads the template, substitutes the input content, calls the API, and writes structured output to the output file.

The batch mode runs workers in parallel (default 4) with automatic backoff on rate limit errors. For a 100-file batch, it typically runs in 30-60 seconds depending on file size and API response time.

When to use a prompt vs. when to write code

The pattern I use for deciding whether a decision goes in code or in a prompt:

Write code if:

The classification is fully enumerable in advance (status == 200)
The inputs are structured and deterministic (dates, numbers, known enum values)
You need the decision to be auditable without an API call
You're running at high volume where API costs matter

Use a prompt if:

The classification requires reading unstructured text
The input format varies in ways you can't predict
Edge cases exist that you haven't mapped yet
Getting it 80% right automatically is better than getting it 0% right manually

The boundary I've found in practice: code handles the deterministic path, prompts handle the long tail. A well-designed pipeline has both.

A real example: automated PR review triage

The most useful pipeline I've built using this pattern: automated triage of incoming GitHub PR review requests.

The problem: I work on a codebase that gets 15-30 PRs per day. Not all of them need deep review — some are docs updates, typo fixes, or dependency bumps. But identifying which ones can be auto-approved vs. which ones need attention requires reading the diff.

The scripts involved:

webhook_receiver.py — receives GitHub webhook, writes PR metadata + diff to a queue directory
prompt_runner.py — classifies each PR using the triage prompt
pipeline_router.py — routes based on classification
notification_sender.py — sends Slack notification with routing decision

The triage prompt:

TRIAGE_PR = """You are a senior developer reviewing pull requests for triage priority.

Given a pull request diff and metadata, classify it as:
- AUTO: safe to auto-approve (docs, formatting, dependency updates with no API changes, comment additions)
- REVIEW: needs human review (logic changes, new features, API changes, security-sensitive code)
- URGENT: needs immediate review (rollback, hotfix, security patch)

Rules:
- If you can't determine the scope from the diff, classify as REVIEW
- If any test files are modified, classify as at least REVIEW
- If any auth, security, or credential-handling code is touched, classify as URGENT

Respond with exactly one word: AUTO, REVIEW, or URGENT.

PR title: {title}
Author: {author}
Files changed: {file_count}
Diff summary:
{diff}
"""

The routing logic:

# pipeline_router.py takes the triage output and acts on it
AUTO   → add "auto-approved" label, merge if checks pass
REVIEW → add to review queue, notify team on next batch
URGENT → page on-call immediately via notification_sender.py

What it does: Reduces manual triage from 20-30 minutes per day to 2-3 minutes. AUTO classifications are almost always right (>95%). URGENT classifications have had 0 false negatives in 6 months — it's conservative on purpose.

The cost question

Running AI calls in a pipeline has a cost. The math that made this viable for me:

Average classification call: ~200 input tokens + 1 output token
At current API pricing: < $0.001 per call
100-file daily batch: < $0.10/day

For most automation pipelines, the cost of AI triage is dominated by the time cost of writing and maintaining the if/elif tree it replaces — which needs to be updated every time a new edge case appears.

The break-even is roughly: if you'd spend more than 10 minutes writing code to handle the edge case, the API call is cheaper.

Combining Part 1 scripts with the AI layer

The scripts most useful as AI pipeline components:

Script	Role in AI pipeline
`http_client.py`	Fetch data for AI processing; handles retries on API rate limits
`file_watcher.py`	Trigger AI triage when new files appear
`webhook_receiver.py`	Accept external events that feed into triage
`queue_processor.py`	Process AI-classified items from a queue
`notification_sender.py`	Send routing decisions and escalations
`retry_runner.py`	Wrap AI calls themselves — handles transient API errors
`cron_wrapper.py`	Schedule batch AI processing jobs

The pipeline pattern:

Collect — http_client.py or webhook_receiver.py or file_watcher.py
Triage — prompt_runner.py with your classification prompt
Route — pipeline_router.py based on triage output
Process — whichever scripts fit the routed path
Notify — notification_sender.py on completion or escalation

You don't always need all five stages. The triage step is what makes it AI-in-the-loop instead of pure script chaining.

Building your first AI triage step

If you have a pipeline that currently uses manual classification or a brittle if tree, here's the migration path:

Identify the judgment call. Where in the pipeline does a human or messy code currently make a decision that requires reading unstructured input?
Define the output enum. What are the possible classifications? 2-4 choices is ideal. More than 6 and the prompt gets unreliable.
Write the prompt. Follow the pattern: role assignment → enumerated choices with definitions → rules for edge cases → required output format (one word). Test it on 10-20 representative examples before deploying.
Add prompt_runner.py to the pipeline. Replace the judgment call with a script call. Log outputs for the first week to validate accuracy.
Tune the rules. After seeing real output, refine the rules section of the prompt to handle edge cases you didn't anticipate. This is faster than updating code.

The prompt library

The prompts above are starting points — designed to be modified for your pipeline's specific enum and rules. The general shape is always the same: role assignment → enumerated choices with definitions → rules for edge cases → required one-word output format.

Once you have a working triage prompt, the marginal cost of adding classification steps to your pipeline is low. Each new decision point is another prompt + another routing branch in pipeline_router.py.

If you're building agent SaaS on top of these pipelines

One thing I keep running into as I expand these pipelines into production services: the pricing model matters as much as the architecture. Pipelines that run on a per-request basis have fundamentally different cost structures than pipelines that run on a per-user seat. If you're building toward a paid product, worth thinking about early.

I wrote a full breakdown of the six pricing models that work for AI agent products — including worked examples, a cost calculator, and the decision framework I actually use:

Pricing Your Python AI Agent SaaS in 2026 →
Six pricing models, Notion cost calculator, worked examples. $39 one-time.

And if you want Part 3 (failure modes, output validation, human-in-the-loop checkpoints) in your inbox when it drops — the list is here:

Subscribe →

What part 3 covers

Part 3 is about failure modes — specifically, what happens when an AI step in your pipeline produces the wrong classification and you don't catch it. It covers: output validation, human-in-the-loop checkpoints for high-stakes decisions, logging patterns for pipeline observability, and how to build a correction loop that improves prompt accuracy over time.

If you're building AI-in-the-loop pipelines right now, leave a comment with what you're working on — I'm actively building from production use cases, and reader questions drive what goes in Part 3. (#discuss)

Why Per-Seat Pricing Breaks AI Agent SaaS (And What Works Instead)

Peyton Green — Sun, 05 Jul 2026 19:08:53 +0000

Why Per-Seat Pricing Breaks AI Agent SaaS (And What Works Instead)

Last month, I realized my AI agent product was underwater.

Customer A: 3 users, $29/month per seat = $87 revenue. Actual cost to serve: $2.

Customer B: 1 user, $29/month = $29 revenue. Actual cost: $30.

Customer C: 2 users, $29/month per seat = $58 revenue. Actual cost: $200.

I was losing $143 a month on Customer C. Per-seat pricing is broken for agent products.

The Core Problem: Cost Doesn't Scale With Seats

Traditional SaaS cost structure:

Hire a support engineer → handle 50 customers
Rent a server → handle 500 customers
Cost per customer goes down as you scale

Per-seat pricing worked because software cost scaled predictably with users.

AI agents break this. Cost is decoupled from headcount.

An agent running simple queries for ten team members might cost $5/month. Another agent running complex analysis for one person might cost $500/month.

The customer with one seat can consume more margin than the customer with ten seats.

Per-seat pricing charges the opposite.

Why This Matters Now

Three reasons the AI agent market is crashing into this problem simultaneously:

1. LLM costs are volatile. An agent that calls 50 LLM tools to answer one question might use $0.10 of inference cost or $50, depending on the model, the question, and the agent's strategy. You can't absorb that variance with a fixed monthly fee.

2. Agent behavior is unpredictable. You can't know upfront how many LLM calls your agent will make. A customer uploads a dataset, asks a question, and suddenly your agent spawns 10 concurrent threads, each making 100 API calls. Per-seat pricing gives you no way to recover that cost.

3. Customers are used to usage-based billing. Everyone pays AWS, Anthropic, OpenAI by usage. They expect the same from agent SaaS. Telling them "it's $29 per seat, unlimited agent calls" feels like you're hiding the cost structure.

The Framework: Six Pricing Models for Agents

I spent a month researching how other builders are solving this. Here are six pricing models that actually work for agent SaaS. Pick one.

Model 1: Per-Token

Charge for every token your agent consumes.

# Simplified example
response = client.chat.completions.create(model="gpt-4", ...)
tokens_used = response.usage.prompt_tokens + response.usage.completion_tokens
customer_bill += tokens_used * 0.000001  # $0.001 per 1K tokens, for example

Pros: Transparent. Customers understand "I pay for what I use."

Cons: Volatile bills. Customers don't know what they'll spend month-to-month.

Best for: Developer tools, anything where customers already pay by API usage.

Model 2: Per-Tool-Call

Charge per agent action (database query, API call, LLM invocation).

def run_agent_action(tool_name, args):
    result = tool.call(args)
    customer_monthly_calls += 1
    customer_bill += 0.01  # $0.01 per call
    return result

Pros: Simple to understand. Encourages efficient agents (fewer calls = cheaper).

Cons: Tool calls have wildly different costs (a Stripe API call ≠ an LLM call).

Best for: Workflow automation, task completion, anything driven by agent autonomy.

Model 3: Per-Task

Charge once per logical task, regardless of internal complexity.

def complete_customer_request(request):
    # Might make 50 LLM calls, 20 tool calls internally
    result = agent.run(request)

    # But you charge once
    customer_monthly_tasks += 1
    customer_bill += 1.00  # $1 per task
    return result

Pros: Customers love this. "It costs $1 to do X." Simple and clear.

Cons: You absorb all the cost variance. If one task costs $0.50 and another costs $50, you need to know your margins.

Best for: Managed services, fixed-scope work.

Model 4: Per-Outcome

Charge only when the agent delivers value.

"$5 per qualified lead," "$50 per completed integration," "$2 per item successfully synced."

Pros: Perfect alignment. You get paid only when the customer wins.

Cons: You have to define and validate "success." High operational overhead. You absorb 100% of cost variance.

Best for: High-trust partnerships, anything with clear, measurable outcomes.

Model 5: Hybrid (Seat + Usage)

Charge a base per-user fee, plus overage for heavy users.

"$29/month per user, includes 5M tokens. After that, $0.30 per 1M tokens."

Pros: Familiar to enterprise customers. Predictable base revenue.

Cons: Requires clear communication. Easy to confuse customers. You still absorb overage variance.

Best for: Migrating from per-seat, or teams with mixed usage.

Model 6: Subscription Tiers (No Usage Tracking)

Forget usage entirely. Charge by feature tier.

Free: basic agent, 1 integration
$29/mo: advanced agent, 5 integrations
$99/mo: unlimited integrations, API access

Pros: Simplest to implement. Customers love predictability.

Cons: You absorb all cost variance. Risky if you don't know your margins.

Best for: Only if you have very tight cost control.

How to Choose

Ask yourself:

Do I understand my customers' agent workload? → Per-task or Per-outcome
Is it highly varied? → Per-token or Per-tool-call
Do customers already understand LLM pricing? → Per-token
Do I want the simplest implementation? → Subscription tiers or Per-task
Do I want to align with customer success? → Per-outcome

I chose Per-tool-call for my product. Customers understand "actions per month." It encourages efficient agent design. It decouples from LLM cost variance.

The Harder Problem: Free Tier

Your free tier is a lead magnet and a cost center.

Bad free tier: "Unlimited free, hope they convert." (They don't. You go broke.)

Good free tier: Clear boundary. Users hit it after 1-2 weeks. Upgrade feels like the obvious choice.

How to size it:

Calculate actual cost to serve one free user: (LLM calls) × (tokens/call) × (cost/token) + (infrastructure spread across users)
Know your customer acquisition cost (CAC): ~$50 for developer products.
Know your payback period math: 2 months at $27/mo margin = $54 revenue covers your CAC.
Size the free tier so users hit the boundary in ~7-14 days.

# Example: If typical user consumes $0.50/week
# And your free tier = $3-6 budget worth
# They'll hit it in 6-12 weeks, you convert them by week 8

class FreeTierManager:
    monthly_token_limit = 10_000_000  # Will take typical user ~1 week to exhaust

    def can_run(self, customer_id):
        usage = get_monthly_usage(customer_id)
        if usage.tokens > self.monthly_token_limit:
            raise QuotaExceeded("Upgrade to continue.")
        return True

Overage: The Surprise Bill Problem

Overage is inevitable. Customer's agent runs faster than expected. Your job: make sure they understand the overage model before they hit it.

Rules:

Show usage in real-time. "You've used 8M of 10M tokens. At your current pace, you'll hit $50 overage this month."
Soft limit, then hard limit. 80% of plan = warning. 100% = slow down, customer must approve overage.
Make pricing obvious upfront. Not in fine print.
Cap surprises. Let customers buy "$50 overage buffer" in advance.

Refunds: Agent Loops and Accidental Bills

Sometimes an agent gets stuck in a loop. Makes 10,000 LLM calls in 10 seconds. Customer gets charged $500.

Customers rightfully expect a refund. Don't be difficult about this.

Automatic refund triggers (no manual review):

Usage spike > 10x daily average
Single hour > 50% of monthly plan
> 1000 calls in < 60 seconds

The Real Lesson

Pricing is not a one-time decision for agent SaaS. It's an ongoing calibration.

Track your numbers:

Cost to serve (per customer)
Margin by customer tier
Free tier conversion rate
Churn by reason

If you're losing money on a cohort, either your pricing is wrong or your product is inefficient. Don't hide this with a bad pricing model — fix it.

I switched from per-seat to per-tool-call. Customer C still costs more, but now I charge them more, too. Margin is reasonable. Customer B isn't underwater anymore.

It took two weeks to build, one week to migrate existing customers (with a grace period), and one month of close monitoring.

Was it worth it? Yes. Per-seat was a timer on going out of business.

Next Steps

If you're building agent SaaS:

Choose a model from the six above
Calculate your actual costs — spend a week on this
Price for 50-70% gross margin — not less
Track for 30 days — does the math work?
Adjust if you're losing money — don't wait

The tooling to implement any of these six models is straightforward. The hard part is knowing your costs and sticking to your margins.

If you're already shipping, and you picked the wrong model — switch. I know switching is painful. But continuing to lose money is worse.

Have you built agent SaaS? What pricing model are you using? What surprised you? Would love to hear what's working and what broke for you.

Go Deeper: The Full Playbook

If this framework was useful, I put together a 30-page playbook covering all six models in detail — with worked Python examples, a Notion cost calculator you can fill in with your own numbers, and a decision framework for choosing the right model for your product.

Pricing Your Python AI Agent SaaS in 2026

Includes the playbook, Notion template, and decision framework. One-time purchase.

Google ADK 2.0 Is Stable — Why That Makes the OpenAI Split Matter More

Peyton Green — Sat, 04 Jul 2026 08:45:03 +0000

Google ADK 2.0 Is Stable — Why That Makes the OpenAI Split Matter More

Last week, Google shipped Agent Development Kit (ADK) 2.0.0 stable, and the same week, a2a-sdk hit 1.0.3 stable. Together, they're a statement: Google has a complete, production-ready agent framework and is willing to build it without OpenAI alignment.

This matters more than it looks because of what it exposes: OpenAI and Google are building agent infrastructure for fundamentally different problems.

The Release: ADK 2.0.0 Stable

Google's jump from 1.29 to 2.0 signals a defined breaking-change boundary. The API surface is stable enough to commit to. You can now build production agents on ADK without prerelease churn risk.

What makes this significant isn't the version bump — it's the timing. ADK 2.0 stable and a2a-sdk 1.0.3 stable landed in the same week. That's coordinated messaging. Google's narrative is clear: "We have a complete, stable, production-ready multi-agent framework."

The Competitive Fault Line: A2A Rejection

Here's what changed the conversation: In March 2026, OpenAI received a 1,200-line pull request implementing Agent-to-Agent (A2A) protocol support in their agents SDK. They declined it.

A2A is Google's answer to multi-agent coordination. It's a standardized protocol for agents to discover, call, and share context with each other. It solves a concrete problem: if you want to compose capabilities across multiple AI agents (and you do — no single agent model is good at everything), how do you do that without reinventing message protocols every time?

OpenAI's "no thanks" wasn't hostile — it's still an open community feature request. But it signals something fundamental: OpenAI and Google are optimizing for different things.

Google is betting on agent-to-agent coordination — a protocol layer that lets you build systems where agents specialize and delegate to each other.

OpenAI is betting on per-agent capability — Sandbox Agents (shipped in the same window) that let a single agent operate a filesystem, run arbitrary code, and persist snapshots.

Both are real problems. They're just orthogonal.

The Framework Landscape: Where You Stand

If you're evaluating frameworks for a production AI agent system, here's what stability means:

Framework	Status	Multi-Agent Story	Interop
Google ADK 2.0.0	Stable	A2A protocol (1.0.3 stable)	A2A-based; open protocol
OpenAI Agents 0.17	Stable	Sandbox Agents + handoff-based	SDK-internal only
LangChain 0.3	Stable	LangGraph composition	Adapter-based for A2A
Microsoft Semantic Kernel	Stable	Plugin-based orchestration	Native A2A support

ADK 2.0.0's stability matters because it means you can adopt the A2A protocol without worrying about breaking changes. If interoperability across agent systems matters to your architecture (and it should — monolithic agents are a bottleneck), ADK gives you a stable foundation.

OpenAI's refusal of A2A doesn't make OpenAI agents bad. Sandbox Agents are genuinely powerful — they let your agent run code, modify files, and operate an environment. For single-agent, high-autonomy workloads, that's valuable. But if you need your agent to collaborate with specialized agents built by your team or third parties, you're back to handoff-based coordination inside the SDK.

What This Means for Your Architecture

Three questions to ask yourself:

1. Do you need single-agent or multi-agent?
If a single agent with sandbox access solves your problem, OpenAI agents work great. If you're building a system where agents specialize (e.g., one reads docs, one retrieves, one reasons, one executes), ADK + A2A gives you a protocol-based way to compose them.

2. Do you need portability?
A2A is a protocol. It means your agents can interoperate with frameworks that speak A2A (Google ADK, LangChain, Semantic Kernel). OpenAI's handoff model is SDK-specific — your orchestration logic lives in OpenAI agent code, not in a portable protocol.

3. Do you need capability stability?
ADK 2.0.0 stable and a2a-sdk 1.0.3 stable mean you can depend on these APIs. OpenAI's agent SDK is stable too, but multi-agent coordination still relies on feature requests and community pressure.

The Code Difference: One Example

Here's how the approaches differ in practice:

With OpenAI agents and handoff:

from openai.agents import Agent

agent_a = Agent(
    model="gpt-4-turbo",
    tools=[retrieve_docs],
)

agent_b = Agent(
    model="gpt-4-turbo",
    tools=[reason_and_execute],
)

# Coordination is SDK-internal
response = agent_a.run("Get docs and ask agent_b to reason about them")
# Agent A has to decide to hand off; agent B has no standard way to discover agent A

With Google ADK + A2A:

from google.genai.adk import Agent
from a2a_sdk import A2AClient

agent_a = Agent(
    name="document_retriever",
    model="gemini-2.0-pro",
    tools=[retrieve_docs],
)

agent_b = Agent(
    name="reasoner",
    model="gemini-2.0-pro",
    tools=[reason_and_execute],
)

# A2A-based discovery and calling
client = A2AClient()
agent_b = client.discover("reasoner")
response = agent_b.call(query="Reason about these docs", context=docs)

The second approach is more portable — agent_b doesn't have to live inside the same process or SDK context as agent_a. It's a network boundary that frameworks can standardize around.

The Strategic Question

This is the moment where framework choice locks you into an architectural commitment. If you're building:

A single, high-autonomy agent: OpenAI agents with Sandbox Agents will give you more capability per agent.
A system of specialized agents: ADK + A2A gives you a protocol layer.
A hybrid approach: LangGraph (LangChain) and Semantic Kernel both support A2A, so you can start with composition and migrate capability where it matters.

Google's stability release says: "We're committed to this direction." OpenAI's rejection of A2A says: "We're optimizing for something else." Both are valid bets. Your architecture decides which one wins for you.

The real shift is that you now have two stable, competing visions for what agent infrastructure should be. ADK 2.0.0 stable didn't change what's possible — it changed what's dependable.

Next: If you're upgrading to newer agent frameworks, watch for breaking changes. OpenAI's 0.13 → 0.17 window introduced two silent breaks that will bite you if you're not careful.

OpenAI Agents SDK 0.13 to 0.17: Three Breaking Changes You Will Hit

Peyton Green — Fri, 03 Jul 2026 18:45:31 +0000

OpenAI Agents SDK 0.13 → 0.17: Three Breaking Changes You Will Hit

The OpenAI Agents SDK moved from 0.13 to 0.17 in five weeks (April 9 – May 19, 2026). That's 19 releases, including three breaking changes. Two of them are silent — they change runtime behavior without raising an error at upgrade time.

If you're running agents on 0.13.x, this is the migration guide you need before you upgrade.

The Three Breaks

Break #1: ModelRefusalError (v0.15.0) — Silent to Loud

The change: Model refusals used to return empty-string output. Now they raise an exception.

What this means: If your code treated refusals as output == "" or checked for empty responses, you were silent-failing gracefully. That behavior changed. A model refusal now raises ModelRefusalError and will crash your agent run unless you handle it.

What you need to do:

from openai.agents import Agent, RunConfig

def handle_refusal(error):
    print(f"Model refused: {error.reason}")
    return "Model refused to respond. Try rephrasing."

agent = Agent(
    model="gpt-4",
    tools=[...],
)

config = RunConfig(
    on_error={
        "model_refusal": handle_refusal,
    }
)

response = agent.run("...", config=config)

Without the handler, you'll see:

ModelRefusalError: model declined to process this request

This is an intentional change — OpenAI wants refusals to be explicit, not silent. But it means your error handling has to change.

Impact: Any agent that doesn't add a refusal handler will crash on a refusal. High-risk if your agents field user input directly.

Break #2: Default Model Changed (v0.16.0) — Silent Model Switch

The change: The default model switched from gpt-4.1 to gpt-5.4-mini.

What this means: If you created an agent without explicitly setting model=, you were running on gpt-4.1. On upgrade to 0.16.0+, that same code silently switches to gpt-5.4-mini.

This isn't just a version bump — gpt-5.4-mini ships with different defaults:

reasoning.effort="none" (GPT-5 default, not OpenAI's historical "medium")
verbosity="low" (more concise output)
Different token economics (cheaper, but different quality/latency tradeoffs)

What you need to do:

Pin your model explicitly before upgrading:

from openai.agents import Agent, ModelSettings

agent = Agent(
    model="gpt-4.1",  # Explicit pin
    model_settings=ModelSettings(
        reasoning_effort="medium",  # Your baseline
    ),
    tools=[...],
)

Or if you want to adopt gpt-5.4-mini intentionally:

agent = Agent(
    model="gpt-5.4-mini",
    model_settings=ModelSettings(
        reasoning_effort="medium",  # If you want reasoning
    ),
    tools=[...],
)

The gotcha: If you have load tests, benchmarks, or quality baselines from 0.13.x, they won't be directly comparable to 0.16.0+ output without an explicit model pin. You'll think performance degraded when really the model changed.

Impact: High — your agents will produce different outputs, and you won't see an error. Your monitoring and quality thresholds need to account for this.

Break #3: Sandbox Source Paths (v0.17.0) — Existing Sandbox Code Only

The change: If you built Sandbox Agents in 0.14.x, the LocalFile.src and LocalDir.src paths must now resolve inside the materialization base_dir unless explicitly granted via Manifest.extra_path_grants.

Absolute paths outside base_dir now fail at runtime.

What this means: This only affects code that's already on 0.14.x sandboxes. If you're upgrading from 0.13.x (which didn't have Sandbox Agents), this doesn't apply to you. If you added sandboxes in 0.14.x, you need to audit your path handling.

What you need to do:

If you're using sandboxes with files outside the sandbox base directory:

from openai.agents import SandboxAgent, Manifest

manifest = Manifest(
    extra_path_grants=[
        "/path/to/external/data",
        "/shared/models",
    ]
)

agent = SandboxAgent(
    model="gpt-4",
    manifest=manifest,
    tools=[...],
)

Or restructure to put everything inside base_dir.

Impact: Low if you're upgrading from 0.13.x. Medium if you're on 0.14.x with sandboxes.

Migration Checklist

Before upgrading to 0.15+:

[ ] Add a model_refusal error handler
[ ] Explicitly pin your model version
[ ] Test with your typical user inputs to catch refusal scenarios

If you're on 0.14.x sandboxes, before upgrading to 0.17.0:

[ ] Audit all LocalFile.src and LocalDir.src paths
[ ] Either move files into base_dir or add extra_path_grants

After upgrading:

[ ] Run your integration tests — the model change will affect output
[ ] Compare output quality against your 0.13.x baseline (pin the old model in parallel if you want a direct comparison)
[ ] Monitor refusal rates — they may spike depending on your use case

What's the Good News?

OpenAI shipped a massive capability in this window: Sandbox Agents (v0.14.0). They let your agent operate a filesystem, run code, persist snapshots, and mount remote storage (S3, GCS, Azure, etc.).

The following 15 releases were OpenAI hardening the sandbox implementation — fixing path security, session integrity, realtime support, and tracing resilience.

This means the SDK is maturing in the direction that matters: capability and reliability. The three breaks are uncomfortable, but they're the cost of OpenAI standardizing behavior (refusals should be loud, models should be explicit, sandboxes should be secure).

Code Example: Safe Upgrade

Here's a pattern that minimizes risk when upgrading:

from openai.agents import Agent, ModelSettings, RunConfig

def create_upgraded_agent(model="gpt-4.1"):
    """Safe agent factory for 0.15+ environments."""

    def handle_refusal(error):
        # Explicit refusal handling
        return f"Cannot proceed: {error.reason}"

    agent = Agent(
        model=model,  # Pinned, no silent switches
        model_settings=ModelSettings(
            reasoning_effort="medium",  # Explicit defaults
        ),
        tools=[...],
    )

    config = RunConfig(
        on_error={
            "model_refusal": handle_refusal,
        },
        max_turns=20,
    )

    return agent, config

# Use it
agent, config = create_upgraded_agent()
response = agent.run("...", config=config)

This pattern:

Pins the model (no surprises)
Handles refusals explicitly (no crashes)
Sets reasonable defaults (explicit config)
Makes it easy to A/B test different models

The Bigger Picture

OpenAI shipped Sandbox Agents and spent 15 releases making them reliable. That's a real bet on capability + stability. The breaking changes are uncomfortable, but they're the cost of standardizing around behavior that should have been the baseline earlier.

If you're upgrading, treat it as a quality improvement with a migration cost — not a regression. The new defaults (gpt-5.4-mini, loud refusals, sandboxes) are better design. You just have to be explicit about it.

Next: If you're building multi-agent systems, watch the framework landscape. Google shipped ADK 2.0.0 stable + A2A 1.0.3 the same week OpenAI shipped this release. Different bets on what agents should do.

LangChain's A2A Integration: Building Multi-Agent Systems in Python Without the Cloud Lock-In

Peyton Green — Tue, 09 Jun 2026 14:00:00 +0000

LangChain released langchain-adk 0.3.14 on March 20, 2026. It's the first mainstream Python agent framework to ship native A2A support — the agent-to-agent coordination protocol that Google, Microsoft, Salesforce, and 100+ enterprises are now treating as infrastructure.

The headline from the release notes is "A2A-first integration." What that means in practice: any LangChain agent can now expose itself as an A2A server and call other A2A-compliant agents, regardless of what framework built them. Your LangChain orchestrator can delegate to a Crew.AI specialist. Your Google ADK agent can call a LangChain specialist. The framework boundary stops mattering.

Note: langchain-adk uses plain asyncio for agent orchestration — it is not built on LangGraph and does not require it. If you're currently using LangGraph, langchain-adk is an alternative path, not an integration layer.

That's the promise. Let's look at what it looks like in code.

What A2A Is (If You're Just Catching Up)

A2A is an open protocol for agent-to-agent task delegation. Version 1.0 dropped March 12, 2026, co-owned by Google and the Linux Foundation, with backing from OpenAI, Microsoft, Salesforce, and ServiceNow. Apache 2.0 license.

The mental model: A2A is to agents what HTTP is to web servers. It gives every agent a common language for sending tasks, reporting progress, and returning results — independent of framework or hosting environment.

If you're already using MCP: MCP handles agent-to-tool coordination — your agent calling APIs, reading files, running code. A2A handles agent-to-agent coordination — your agents delegating tasks to each other. They're complementary layers. Google's own framing from the v1.0 announcement positions them exactly that way. You'll end up using both.

Every A2A agent publishes an "Agent Card" at /.well-known/agent-card.json. That card lists its capabilities, input schema, and endpoint. Any A2A client that reads the card knows what it can do and how to talk to it.

If you want the hands-on version: Build Your First A2A Agent Pair in Python (15 Minutes, No Cloud Required) covers the base protocol from scratch.

What Changed in langchain-adk 0.3.14

Before this release, using A2A with LangChain required custom server code — you'd wrap a chain or agent in a FastAPI app, manually implement the A2A task endpoint, write the Agent Card JSON by hand. It worked, but it was 200+ lines of boilerplate that you had to maintain.

0.3.14 ships three additions:

1. A2AServer wrapper for LangChain agents

Wraps any LangChain agent in an A2A-compliant HTTP server. One import, one class, done.

2. A2AAgent for calling external agents

A LangChain-native class that wraps a remote A2A agent as a callable component. Your orchestrator agent calls remote A2A agents exactly like it calls local tools.

3. Auto-generated Agent Cards

Agent Cards are generated from the agent's name, description, and tool list. No manual JSON authoring.

Building a Multi-Agent Pipeline with langchain-adk

Here's the pattern: one orchestrator agent that coordinates tasks, and two specialist agents that handle specific work. All three speak A2A.

Setup:

pip install "langchain-adk[a2a]" langchain-openai uvicorn

Specialist agents (run these in separate processes):

# code_reviewer.py
import uvicorn
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_adk.a2a import A2AServer, AgentSkill
from langchain_adk.agents.llm_agent import LlmAgent
from langchain_adk.sessions import InMemorySessionService

llm = ChatOpenAI(model="gpt-4o-mini")

@tool
def review_code(code: str) -> str:
    """Review Python code for bugs, style issues, and improvement opportunities."""
    response = llm.invoke(f"Review this Python code:\n\n{code}\n\nList issues and improvements.")
    return response.content

agent = LlmAgent(
    name="code-reviewer-agent",
    llm=llm,
    tools=[review_code],
    description="Reviews Python code for bugs, style issues, and performance problems",
)

# Wrap it as an A2A server
server = A2AServer(
    agent=agent,
    session_service=InMemorySessionService(),
    app_name="code-reviewer",
    url="http://localhost:8001",
    skills=[
        AgentSkill(
            id="review",
            name="Code Review",
            description="Reviews Python code for bugs, style issues, and performance problems",
            tags=["python", "review"],
        )
    ],
)

# Serve with uvicorn — Agent Card published at /.well-known/agent-card.json
app = server.as_fastapi_app()
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)

# test_generator.py
import uvicorn
from langchain.tools import tool
from langchain_adk.a2a import A2AServer, AgentSkill
from langchain_adk.agents.llm_agent import LlmAgent
from langchain_adk.sessions import InMemorySessionService
# ... similar @tool definition for test generation ...

agent = LlmAgent(name="test-generator-agent", llm=llm, tools=[generate_tests], description="Generates pytest tests")
server = A2AServer(
    agent=agent,
    session_service=InMemorySessionService(),
    app_name="test-generator",
    url="http://localhost:8002",
    skills=[AgentSkill(id="test", name="Test Generator", description="Generates pytest tests", tags=["python", "testing"])],
)
app = server.as_fastapi_app()
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8002)

Orchestrator agent (calls the specialists):

# orchestrator.py
import asyncio
from langchain_openai import ChatOpenAI
from langchain_adk import A2AAgent
from langchain_adk.agents.react_agent import ReActAgent
from langchain_adk.tools.agent_tool import AgentTool
from langchain_adk.sessions import InMemorySessionService
from langchain_adk.runner import Runner

llm = ChatOpenAI(model="gpt-4o-mini")

# Connect to remote A2A specialists — url is the base server URL, not the agent-card path
code_reviewer = A2AAgent(
    name="code-reviewer",
    url="http://localhost:8001",
    description="Reviews Python code for bugs, style issues, and performance problems",
)
test_generator = A2AAgent(
    name="test-generator",
    url="http://localhost:8002",
    description="Generates pytest tests for Python functions",
)

# AgentTool wraps BaseAgent as a BaseTool so ReActAgent can call it via tool-use
orchestrator = ReActAgent(
    name="orchestrator",
    llm=llm,
    tools=[AgentTool(code_reviewer), AgentTool(test_generator)],
)

async def main():
    session_service = InMemorySessionService()
    runner = Runner(agent=orchestrator, app_name="pipeline", session_service=session_service)
    session = await runner.get_or_create_session(user_id="user", session_id="session-1")

    async for event in runner.astream(
        user_id="user",
        session_id="session-1",
        new_message="Review this code and generate tests for the functions that pass review:\n\n" + code,
    ):
        if hasattr(event, "text") and event.text:
            print(event.text)

asyncio.run(main())

The orchestrator never imports the specialist code. A2AAgent discovers the remote agent's endpoint at runtime and delegates via the A2A protocol. Swap out a specialist for a different implementation — the orchestrator adapts automatically.

The Framework Portability Test

The real value isn't just LangChain-to-LangChain coordination. It's cross-framework.

Because your specialist agents speak A2A, an AutoGen orchestrator can call them. A CrewAI pipeline can delegate to them. A custom Python agent built with zero framework can call them using the base a2a-sdk client.

# Additional dependencies for the portability section only
pip install "a2a-sdk==0.3.25" httpx

# calling your LangChain specialist from a non-LangChain orchestrator
import asyncio
import httpx
from a2a.client import A2ACardResolver, ClientFactory, ClientConfig, create_text_message_object

BASE_URL = "http://localhost:8001"

async def call_specialist(code: str) -> str:
    async with httpx.AsyncClient() as http:
        # Resolve the Agent Card
        resolver = A2ACardResolver(httpx_client=http, base_url=BASE_URL)
        card = await resolver.get_agent_card()

        # Create a client and send the task
        factory = ClientFactory(config=ClientConfig(httpx_client=http))
        client = factory.create(card)

        message = create_text_message_object(content=f"Review this code: {code}")
        async for event in client.send_message(message):
            if hasattr(event, "parts"):
                for part in event.parts:
                    if hasattr(part.root, "text"):
                        return part.root.text
    return ""

asyncio.run(call_specialist(code))

That's the protocol doing its job. The LangChain specialist doesn't know it's being called from a non-LangChain client.

Observability: Tracing Multi-Agent Systems with LangSmith

langchain-adk 0.3.14 hooks into LangChain's standard callback system, so any callback handler you already use works with A2A agents. LangSmith is the most common choice:

from langsmith import Client
from langchain.callbacks import LangChainTracer

tracer = LangChainTracer(project_name="multi-agent-pipeline")

# Pass it to each agent — both specialists and orchestrator
orchestrator = AgentExecutor(agent=agent, tools=tools, callbacks=[tracer])

Because the tracer spans all agents in a task, you get a single trace tree even when the orchestrator delegates to remote specialists. Langfuse, Weights & Biases, or any BaseCallbackHandler subclass works the same way — langchain-adk doesn't require any specific observability provider.

For local development, print callbacks are enough. For production — when an orchestrator delegates to three specialists and one returns a bad result — trace stitching is the difference between "something failed" and "specialist agent 2 hallucinated on this specific input at 14:23 UTC."

When to Use This Pattern

Use A2A-based agent delegation when:

Different agents need different models (fast model for triage, expensive model for complex reasoning)
Different agents need different tool sets that you don't want to expose to each other
You want framework portability — the ability to swap out an agent implementation without touching the orchestrator
You're scaling to multiple concurrent agents that need to coordinate on shared tasks

Don't reach for it when you have a simple chain that runs sequentially. A2A coordination adds network hops and complexity. For a single-agent flow, keep it simple.

The Bigger Picture

langchain-adk 0.3.14 is significant because LangChain is the Python framework with the most production deployments in the AI developer community. When the most widely-used framework ships A2A-first, the protocol crosses a threshold from "interesting spec" to "thing you'll encounter in the wild."

MCP and A2A are converging on being the two required protocols for production agent systems: MCP handles agent-to-tool (your agent calling APIs, reading files, running code), A2A handles agent-to-agent (your agents delegating to each other). Frameworks that support both are frameworks that can build anything.

If you're building multi-agent Python systems and haven't integrated A2A yet, this release is a reasonable entry point. The LangChain wrapper handles the boilerplate. The protocol handles the coordination.

The AI Dev Toolkit includes five Python scripts that build on these patterns — including an orchestrator template, a multi-agent review pipeline, and a batch processor that distributes work across specialist agents. Get the toolkit here.

google-adk 2.0 Is Now Stable: Workflow Runtimes, Breaking Changes, and How to Migrate

Peyton Green — Tue, 02 Jun 2026 21:00:06 +0000

What google-adk 2.0 Changes

1. The Workflow Runtime (The Big One)

The biggest structural change in 2.0 is a new graph-based workflow execution engine. In 1.x, agent orchestration was largely LLM-driven — you defined tools and let the model decide when to use them. In 2.0, you can define deterministic workflows that execute regardless of what the LLM decides.

from google.adk.workflow import Workflow, node

# Define workflow nodes using the @node decorator
@node
async def validate_input(state: dict) -> dict:
    return {**state, "validated": True}

@node
async def call_external_api(state: dict) -> dict:
    # Fan-out: triggers parallel downstream nodes
    return {**state, "api_called": True}

# Compose into a Workflow
workflow = Workflow(
    nodes=[validate_input, call_external_api]
)

This is direct competition with LangGraph's StateGraph. The key difference: ADK's Workflow Runtime is Google-native, has built-in A2A protocol support, and handles the session/state persistence layer automatically.

Supported patterns:

Sequential pipelines
Fan-out / fan-in (parallel subagent execution)
Loops with retry logic
Dynamic routing based on node output
Human-in-the-loop breakpoints
Nested workflows

2. The Event Model Is Completely Different

In 1.x, data passed between agent components through tool call results and prompt context. In 2.0, Events are the primary data primitive. Everything flows as events.

from google.adk.sessions import InMemorySessionService
from google.adk.sessions.state import State

# State is managed through the session service, not through events
session_service = InMemorySessionService()
session = await session_service.create_session(app_name="my_app", user_id="user_1")

# Read and write state through the session
async def my_node(state: dict) -> dict:
    user_context = state.get("user_context", {})
    return {**state, "processed": True, "context": user_context}

The state field in an Event persists automatically across workflow nodes — you don't manage it manually. This is the mechanism that replaces 1.x session management.

3. The Task API for Agent-to-Agent Delegation

2.0 introduces a structured Task API for multi-agent systems. Instead of agents calling tools that happen to invoke other agents, you now have explicit task delegation semantics:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner

# Coordinator agent that delegates to sub-agents
coordinator = LlmAgent(
    name="coordinator",
    model="gemini-2.0-flash",
    sub_agents=[researcher_agent, writer_agent],
    instruction="Coordinate research and writing tasks."
)

# Run with explicit delegation through the runner
runner = Runner(agent=coordinator, app_name="my_app")

This integrates with the A2A protocol, so ADK agents can delegate tasks to non-ADK agents over the standard A2A wire format.

What Breaks in 2.0

Storage Incompatibility (Critical)

Do not use ADK 2.0 storage with ADK 1.x databases. This isn't a "things might break" warning — Google explicitly flags this as data loss territory. The session schema, memory systems, and evaluation data storage are all incompatible between versions.

If you're running 1.x in production, you need to:

Back up your session data before any migration work
Run 1.x and 2.0 projects in completely separate storage environments
Plan a migration window, not an in-place upgrade

Agent API Changes

The top-level agent API has breaking changes across the board. In 1.x, agents were structured around tool calls and hierarchical execution. In 2.0, the entire execution model shifts to graph-based workflows with explicit node definitions and event routing. Key API breaks:

Agent initialization: 1.x Agent() constructor signature changed. 2.0 agents must be wrapped in or deployed as WorkflowRuntime nodes.
Tool definition: 1.x @tool decorators are superseded by @WorkflowNode patterns. Tool invocation is now event-routed, not LLM-driven.
Session state: 1.x session objects with .state attributes replaced by Event-based state persistence. All state must flow through Event.with_state().
Response format: 1.x agent responses (direct returns or text) now return structured Event objects with output and node_info fields.

The general pattern is that 1.x agents that used the tool-centric API will need to be restructured around the Event model and WorkflowRuntime graph orchestration.

Session Schema

Session handling is rebuilt around the Event model. Any code that directly manipulated session state in 1.x will need to be rewritten.

Python Version Requirement

2.0 requires Python 3.11+. If you're on 3.10 or earlier, you'll need to upgrade your environment before you can use 2.0.0.

The Security Note: LiteLLM

2.0.0a2 explicitly pins LiteLLM to version 1.82.6, excluding 1.82.7 and 1.82.8. Those two versions were compromised in a supply chain incident. If your 1.x setup uses LiteLLM and you haven't checked your pinned version, check it now — independently of any ADK migration.

# Check your current LiteLLM version
pip show litellm | grep Version

# If you see 1.82.7 or 1.82.8, upgrade immediately
pip install "litellm==1.82.6"

Migration Strategy

Don't migrate during a sprint. Here's the sequence that makes sense:

Step 1: Audit your 1.x footprint

# Find all ADK imports in your codebase
grep -r "from google.adk" . --include="*.py"
grep -r "import google.adk" . --include="*.py"

Step 2: Isolate storage
Stand up a separate dev environment with fresh storage before touching any 2.0 code. Your 1.x session data and your 2.0 dev environment should never share a database.

Step 3: Install 2.0.0 stable
Standard pip install google-adk now pulls 2.0.0 by default. Explicit pin optional but recommended:

pip install "google-adk==2.0.0"

Or update if you're already on a 2.0 alpha:

pip install --upgrade "google-adk==2.0.0"

Step 4: Port your agents to the Event model
The workflow is: identify what state your agents pass around → convert to Event state fields → rebuild orchestration logic using WorkflowRuntime instead of tool chains.

Step 5: Update your A2A integrations
If you're using the A2A protocol, the Task API in 2.0 gives you a cleaner integration path than the 1.x approach. Worth rewriting these deliberately rather than patching.

When to Migrate

2.0.0 is now stable (released May 19, 2026). The migration window is open. Production workloads on 1.x should be evaluated for migration readiness — don't rush, but don't treat this as "far away" anymore. Most teams should plan a migration within the next 2-4 weeks depending on code complexity and storage requirements.

What This Means for Your Tooling

The Workflow Runtime makes ADK 2.0 a direct competitor to LangGraph for Python agent orchestration. The big difference: ADK is Google-native with built-in Vertex AI, Cloud Run, and A2A integration. If you're already deep in Google Cloud, this is worth evaluating seriously.

If you're not on Google Cloud, LangGraph still has a more mature ecosystem and the new A2A Task API in ADK 2.0 doesn't change that calculus much.

For evaluating any major agent framework migration, the AI Dev Toolkit has 30+ prompts specifically for agent architecture decisions — framework comparisons, migration planning, multi-agent design review. Cheaper than an afternoon of context-switching.

Summary

What changed	Impact
Workflow Runtime (graph-based)	New architecture for deterministic agent flows — as of 2.0.0 stable
Event model	All data now flows as Events — major rewrite of state management — as of 2.0.0 stable
Task API	Structured agent-to-agent delegation, A2A-native — as of 2.0.0 stable
Storage	1.x and 2.0 storage are incompatible — separate environments required — confirmed in 2.0.0 stable
Session schema	Rebuilt — 1.x session code needs rewriting — confirmed in 2.0.0 stable
Python 3.11+ required	Upgrade your environment if on 3.10 or earlier — confirmed in 2.0.0 stable
LiteLLM security	2.0.0 stable excludes 1.82.7-1.82.8 (compromised). Check your pins immediately if running < 1.82.6.

Spotted something that changed between this article and current ADK 2.0 stable? Drop it in the comments — this space moves fast.

Python's Toolchain Anxiety Problem (And How to Think About It)

Peyton Green — Tue, 02 Jun 2026 14:00:04 +0000

Something happened in the Python ecosystem in March 2026 that's worth thinking through carefully.

In the span of two weeks, two of the most-used Python developer tools either changed their access model or changed owners: LocalStack sunset its free tier and required account registration, and OpenAI announced it's acquiring Astral — the company behind uv, ruff, and ty.

Neither event is a catastrophe. But together, they reveal a pattern worth naming.

The Pattern

Here's how it goes:

A developer tool solves a real problem, better than anything else
Developers adopt it. Sometimes enthusiastically, sometimes because it became the default
The tool grows. The company behind it needs a business model
The business model change happens — subscription, account gate, acquisition, licensing shift
Developers are now dependent on infrastructure they don't fully control

This isn't new. Terraform → OpenTofu. Docker Desktop. GitHub → Microsoft. Elasticsearch. The list is long. What's new is the pace, and the depth of the dependency.

uv isn't a nice-to-have. For many Python projects in 2026, it's the package manager. ruff has replaced entire linting stacks. These tools are upstream of everything. When they move under corporate control, the question isn't whether to care — it's whether you have an exit if the governance terms turn hostile.

What LocalStack Actually Did

Let's be specific about LocalStack, because the community reaction was complicated.

LocalStack added an auth token requirement to localstack/localstack:latest. This means: you need an account to run LocalStack in CI. The free tier still exists. For individual developers, the change is minor.

But for teams, for OSS projects, for anyone running ephemeral CI containers that spun up LocalStack 500 times a day without a second thought — the friction is real. It's not a cost barrier (free tier exists), it's an account barrier. You now have a dependency relationship with LocalStack.com that didn't exist before.

The community's reaction wasn't disproportionate. It was pattern recognition. "We've seen this before. First it's accounts, then it's rate limits, then it's pricing tiers, then we're migrating again in two years."

LocalStack has since announced a partial rollback — a Hobby tier, unlimited CI credits with fair-use limits. That's a good response to community pressure. But the account requirement remains. The dependency relationship is still there.

The developers who switched to Moto or Floci immediately weren't being dramatic. They were protecting their time.

What Astral Actually Did (So Far: Nothing)

The Astral situation is different — and it's worth being precise here, because the community anxiety ran ahead of the actual events.

OpenAI announced an intent to acquire Astral. At the time of this writing, the acquisition is pending regulatory close. The governance terms haven't been disclosed. No licensing changes have happened. uv and ruff are still Apache 2.0, still community-developed, still exactly what they were last week.

The community anxiety wasn't about what happened. It was about what might happen. That's a reasonable thing to be anxious about — OpenAI has incentives to tie toolchain improvements to Codex workflows, or to slow the open roadmap, or to introduce Codex-specific optimizations. These are possibilities, not certainties.

The HN thread had 757 points and 475 comments within hours. Simon Willison wrote analysis the same day. That's a community that's had its threat model primed by prior events. They've lived through Terraform, Docker Desktop, Elasticsearch. They're not going to wait to see how this plays out before thinking about contingencies.

That's also healthy. The contingency for ruff probably looks like a maintained fork, the same way the contingency for Terraform became OpenTofu. No one needs to run to an alternative today. But knowing the fork exists, and following it, is sensible insurance.

What This Isn't

This isn't an argument that OSS companies are bad or that developers shouldn't pay for software.

LocalStack built something genuinely valuable. They have engineers to pay and infrastructure to run. The Hobby tier is real — you can run LocalStack for free. The ask is modest.

Astral built uv from scratch and made Python package management fast for the first time in years. They deserve to have a business model. OpenAI paying them for that is a reasonable outcome.

The tension isn't companies wanting to make money. The tension is: when foundational developer tooling sits inside corporate ownership, the roadmap is not yours, and the terms can change.

What to Actually Do About It

Three things, in order of priority:

1. Know your dependencies' governance

For your most critical tooling — the stuff that would cost you a week to replace — do you know who owns it? Under what license? What happens if they're acquired? This doesn't require action, just awareness. A 30-minute inventory is worth doing.

2. Watch for account relationships that didn't exist before

Account gates are the canary. The transition from "download and run" to "download, register, and authenticate" is meaningful. It's not always bad — but it is a change in the dependency relationship. When you notice it, flag it.

3. Favor tools that don't require ongoing trust

Where you have a choice, weight towards tools you can own, fork, or run offline. One-time purchases over subscriptions where the value is equivalent. Tools with neutral governance (Apache Software Foundation, Linux Foundation, Agentic AI Foundation) over tools with a single corporate owner. Not as a religion — as a tiebreaker.

This doesn't mean boycotting SaaS or refusing accounts. It means being deliberate about where you have dependencies that require ongoing trust relationships.

The Python Community Is Actually Good at This

Here's the underreported part of both stories: the Python community responded fast and well.

LocalStack sunset their free tier on March 23. By March 24, Moto — the Python-native AWS mock library with 200+ service coverage and no Docker, no account, no auth token — had become the canonical recommendation in the community. Floci, a drop-in LocalStack replacement that's MIT licensed, hit 1,500+ GitHub stars in five days.

The community didn't wait. They assessed the situation, identified the alternatives, and moved. That's the ecosystem working as it should.

The Astral acquisition will play out over months. If the governance terms are favorable, nothing changes. If the terms signal commercial capture of the roadmap, someone will fork. They'll call it OpenUV or something equally unsubtle. It'll have 10,000 GitHub stars within a week and be indistinguishable from uv within a year.

The Python community has been through this. They know what to do.

One More Thing

The frustrating part of toolchain anxiety isn't the individual events. It's the cumulative tax on developer attention.

Every time a tool changes ownership or access model, someone has to evaluate it, file the mental note, update the team, maybe update the docs. That's time that doesn't go into building things. The hidden cost of corporate capture of foundational tooling is the aggregate cognitive overhead it generates across millions of developers.

That's the thing worth staying angry about — not any specific company's business decisions, but the ecosystem norm where "we built the thing everyone depends on, now we need to change the terms" is just expected behavior.

If you're building developer tools: the trust you earn by being predictable and transparent about your model is worth more than the friction you save by not communicating. LocalStack would have had a better March if they'd published their monetization timeline six months earlier. Astral will have a better acquisition if OpenAI publishes clear governance terms before regulatory close, not after.

Developers are not angry about business models. They're angry about surprises.

What We Build With

For what it's worth: everything I publish in this series runs locally, requires no account, no subscription, no auth token. The code in the Automation Cookbook is code you own — you can fork it, modify it, run it forever without a renewed relationship with anyone.

That's not a moral position. It's a practical one. The tools that earn the most durable trust are the ones that don't require ongoing trust to use.

If you found this useful, the AI Dev Toolkit is 47 curated prompts, scripts, and patterns for Python developers — $29, one-time purchase, no subscription. The Automation Cookbook is 30 production-ready Python automation scripts — $39, same terms.

Previous in this series: Fix Your LocalStack Auth Errors in 5 Minutes (testcontainers + Floci) | Build Your First A2A Agent Pair in Python

Async Python for AI Applications: Patterns That Don't Break Under Load

Peyton Green — Tue, 26 May 2026 21:00:02 +0000

The first async AI application most Python developers write looks like this:

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def summarize(text: str) -> str:
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": f"Summarize: {text}"}]
    )
    return response.content[0].text

async def main():
    results = await asyncio.gather(
        summarize(doc1),
        summarize(doc2),
        summarize(doc3),
    )

It works. Then you run it against 500 documents and get a mix of rate limit errors, connection timeouts, and partial results. Some tasks complete, some silently fail, and you have no idea which.

This post covers the async patterns that actually hold under real load: bounded concurrency, retry with backoff, result collection with error isolation, and cancellation.

The problem with unbounded gather

asyncio.gather(*[summarize(doc) for doc in docs]) fires every task simultaneously. With 500 documents, that's 500 concurrent API connections. Three things happen:

Rate limit errors. Most AI APIs have per-minute token limits. 500 simultaneous requests will hit them instantly.
Connection pool exhaustion. The default httpx connection pool that backs the Anthropic SDK has a default limit of 100 connections. Beyond that, requests queue or fail.
Error propagation. asyncio.gather by default raises the first exception and cancels remaining tasks. One bad document kills the batch.

The fix is a semaphore.

Bounded concurrency with asyncio.Semaphore

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()
semaphore = asyncio.Semaphore(10)  # max 10 concurrent API calls

async def summarize_bounded(text: str) -> str:
    async with semaphore:
        response = await client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=512,
            messages=[{"role": "user", "content": f"Summarize: {text}"}]
        )
        return response.content[0].text

async def process_batch(docs: list[str]) -> list[str]:
    tasks = [summarize_bounded(doc) for doc in docs]
    return await asyncio.gather(*tasks)

The semaphore acts as a queue — 10 tasks run concurrently, the rest wait. Rate limit errors drop significantly. Connection pool stays healthy.

Tuning the semaphore value: Start at 5 for development. For production, tune to floor(rate_limit_per_minute / avg_seconds_per_call / 60). If your API allows 60,000 tokens/minute and each call uses ~1,000 tokens and takes ~1 second, 60 concurrent calls is theoretically safe — use 40-50 to leave headroom.

Retry with exponential backoff

Rate limit errors still happen even with bounded concurrency — you share quota with other processes, quotas vary by time of day, and the API occasionally returns 529s. A retry decorator:

import asyncio
import random
import logging
from functools import wraps
from anthropic import RateLimitError, APIStatusError

logger = logging.getLogger(__name__)

def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except RateLimitError:
                    if attempt == max_retries:
                        raise
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    logger.warning(f"Rate limit hit, retry {attempt + 1}/{max_retries} in {delay:.1f}s")
                    await asyncio.sleep(delay)
                except APIStatusError as e:
                    if e.status_code in (500, 502, 503, 529):
                        if attempt == max_retries:
                            raise
                        delay = base_delay * (2 ** attempt)
                        await asyncio.sleep(delay)
                    else:
                        raise  # 400s, auth errors — don't retry
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=2.0)
async def summarize_with_retry(text: str) -> str:
    async with semaphore:
        response = await client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=512,
            messages=[{"role": "user", "content": f"Summarize: {text}"}]
        )
        return response.content[0].text

Key details:

Exponential backoff (base_delay * 2^attempt) with jitter (random.uniform(0, 1)) prevents thundering herd — all retrying tasks don't fire at the same moment
Only retry on rate limits (429) and transient server errors (500, 502, 503, 529)
Don't retry 400-range errors — bad requests won't succeed on retry and you'll waste quota

Error isolation in batch processing

asyncio.gather(*tasks) propagates the first exception. For batch processing where partial success is acceptable, use return_exceptions=True:

from dataclasses import dataclass
from typing import Any

@dataclass
class BatchResult:
    index: int
    success: bool
    value: str | None
    error: Exception | None

async def process_batch_isolated(docs: list[str]) -> list[BatchResult]:
    tasks = [summarize_with_retry(doc) for doc in docs]
    raw_results = await asyncio.gather(*tasks, return_exceptions=True)

    results = []
    for i, result in enumerate(raw_results):
        if isinstance(result, Exception):
            logger.error(f"Doc {i} failed: {type(result).__name__}: {result}")
            results.append(BatchResult(index=i, success=False, value=None, error=result))
        else:
            results.append(BatchResult(index=i, success=True, value=result, error=None))

    failed = sum(1 for r in results if not r.success)
    logger.info(f"Batch complete: {len(results) - failed}/{len(results)} succeeded")
    return results

With return_exceptions=True, exceptions are returned as values in the results list rather than raised. You decide what to do with failures: log and continue, re-queue for retry, write to a dead-letter queue, or raise.

Progress tracking for long batches

For batches that take minutes, you want progress updates without blocking:

import asyncio
from tqdm.asyncio import tqdm

async def process_with_progress(docs: list[str]) -> list[BatchResult]:
    semaphore = asyncio.Semaphore(10)

    async def process_one(i: int, doc: str) -> BatchResult:
        async with semaphore:
            try:
                result = await summarize_with_retry(doc)
                return BatchResult(index=i, success=True, value=result, error=None)
            except Exception as e:
                return BatchResult(index=i, success=False, value=None, error=e)

    tasks = [process_one(i, doc) for i, doc in enumerate(docs)]

    results = []
    async for result in tqdm.as_completed(tasks, desc="Processing docs", total=len(docs)):
        results.append(await result)

    # Restore original order
    results.sort(key=lambda r: r.index)
    return results

tqdm.asyncio.tqdm.as_completed wraps asyncio.as_completed with a progress bar. Results arrive in completion order, so sort by index at the end if you need original ordering.

Timeouts and cancellation

AI API calls can hang. The AsyncAnthropic client has a default timeout, but you might want stricter control:

async def summarize_with_timeout(text: str, timeout: float = 30.0) -> str:
    try:
        async with asyncio.timeout(timeout):
            async with semaphore:
                response = await client.messages.create(
                    model="claude-sonnet-4-6",
                    max_tokens=512,
                    messages=[{"role": "user", "content": f"Summarize: {text}"}]
                )
                return response.content[0].text
    except asyncio.TimeoutError:
        logger.warning(f"Timeout after {timeout}s for doc (len={len(text)})")
        raise

asyncio.timeout (Python 3.11+) is cleaner than asyncio.wait_for — it raises TimeoutError and properly cancels the underlying task. Set timeout based on your max_tokens and expected model latency, not a round number.

For Python < 3.11, use asyncio.wait_for:

result = await asyncio.wait_for(
    client.messages.create(...),
    timeout=30.0
)

Putting it together: a production batch processor

import asyncio
import logging
from dataclasses import dataclass
from typing import Callable, Awaitable
from anthropic import AsyncAnthropic, RateLimitError, APIStatusError

logger = logging.getLogger(__name__)

@dataclass
class TaskResult[T]:
    index: int
    success: bool
    value: T | None
    error: Exception | None

class BatchProcessor[T]:
    """Bounded-concurrency batch processor with retry and error isolation."""

    def __init__(
        self,
        concurrency: int = 10,
        max_retries: int = 3,
        timeout: float = 60.0,
    ):
        self.semaphore = asyncio.Semaphore(concurrency)
        self.max_retries = max_retries
        self.timeout = timeout

    async def _run_with_retry(
        self, fn: Callable[..., Awaitable[T]], *args, **kwargs
    ) -> T:
        for attempt in range(self.max_retries + 1):
            try:
                async with asyncio.timeout(self.timeout):
                    async with self.semaphore:
                        return await fn(*args, **kwargs)
            except RateLimitError:
                if attempt == self.max_retries:
                    raise
                delay = 2.0 * (2 ** attempt) + random.uniform(0, 1)
                await asyncio.sleep(delay)
            except APIStatusError as e:
                if e.status_code >= 500 and attempt < self.max_retries:
                    await asyncio.sleep(2.0 * (2 ** attempt))
                else:
                    raise

    async def process(
        self,
        fn: Callable[..., Awaitable[T]],
        items: list,
    ) -> list[TaskResult[T]]:
        async def run_one(i: int, item) -> TaskResult[T]:
            try:
                value = await self._run_with_retry(fn, item)
                return TaskResult(index=i, success=True, value=value, error=None)
            except Exception as e:
                logger.error(f"Item {i} failed permanently: {e}")
                return TaskResult(index=i, success=False, value=None, error=e)

        tasks = [run_one(i, item) for i, item in enumerate(items)]
        results = await asyncio.gather(*tasks, return_exceptions=False)
        results.sort(key=lambda r: r.index)

        success_count = sum(1 for r in results if r.success)
        logger.info(f"Batch: {success_count}/{len(results)} succeeded")
        return results


# Usage
async def main():
    import random
    client = AsyncAnthropic()
    processor = BatchProcessor(concurrency=10, max_retries=3)

    async def summarize(doc: str) -> str:
        response = await client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=512,
            messages=[{"role": "user", "content": f"Summarize: {doc}"}]
        )
        return response.content[0].text

    docs = ["doc text here"] * 100
    results = await processor.process(summarize, docs)

    successes = [r.value for r in results if r.success]
    failures = [r for r in results if not r.success]
    print(f"Processed {len(successes)} docs, {len(failures)} failed")

When to use threads instead of async

Async works when the bottleneck is I/O wait — network calls, file reads, database queries. It doesn't help when the bottleneck is CPU.

If you're doing heavy post-processing on AI outputs (parsing, classification, regex over large texts), use asyncio.to_thread to run that work in a thread pool without blocking the event loop:

import asyncio
import re

def extract_entities(text: str) -> list[str]:
    # CPU-bound text processing — runs in thread pool
    pattern = re.compile(r"\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b")
    return pattern.findall(text)

async def analyze_with_extraction(doc: str) -> tuple[str, list[str]]:
    # AI call — async I/O
    summary = await summarize_with_retry(doc)

    # CPU work — offload to thread pool
    entities = await asyncio.to_thread(extract_entities, summary)

    return summary, entities

asyncio.to_thread runs the sync function in the default ThreadPoolExecutor and awaits the result. The event loop remains unblocked. For truly heavy CPU work (model inference, large numpy operations), use ProcessPoolExecutor instead.

A checklist for async AI applications

asyncio.Semaphore on every batch — never unbounded gather against external APIs
Exponential backoff with jitter on rate limit and 5xx errors
return_exceptions=True on gather for batch processing — let failures be values, not raised exceptions
Timeout on every API call — never trust external latency to be bounded
asyncio.to_thread for CPU-bound post-processing — keep the event loop clear

The async patterns in the AI Dev Toolkit include prompt templates for generating asyncio batch processors, retry decorators, and progress-tracked pipelines from function signatures — so you're not writing boilerplate from scratch each time you build a new AI workflow.

Structured LLM Outputs with Pydantic v2: Stop Parsing Freeform JSON and Start Typing Your AI

Peyton Green — Tue, 19 May 2026 14:00:07 +0000

The biggest source of subtle bugs in AI applications isn't the model — it's the gap between what you asked for and what you got.

You prompt for {"score": 8, "issues": ["missing error handling"]} and you get {"score": "8/10", "issues": "missing error handling"}. Both are technically valid JSON. One breaks your downstream code. Neither triggers an exception until hours later when you're wondering why the aggregation is wrong.

Pydantic v2 eliminates this class of bugs. Here's how to structure your LLM outputs so type errors are caught at the boundary, not buried in production.

The problem with freeform JSON parsing

Most developers start here:

import json
from anthropic import Anthropic

client = Anthropic()

def analyze_code(code: str) -> dict:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"Analyze this code and return JSON with: severity (int 1-10), issues (list of strings), has_security_risk (bool).\n\n{code}"
        }]
    )
    return json.loads(response.content[0].text)

This fails in three ways you won't notice until production:

Type coercion silently wrong. The model returns "severity": "8" instead of 8. json.loads parses it as a string. Your downstream severity > 7 comparison evaluates to False for every input.
Missing fields. The model occasionally omits has_security_risk when it seems obvious from context. KeyError three calls in, two hours into a batch job.
Schema drift. You update the prompt. The model starts returning an extra field. Your downstream code ignores it. A week later you realize the data you've been storing is inconsistent.

The Pydantic v2 fix

Define your output schema first:

from pydantic import BaseModel, Field, field_validator
from typing import Annotated

class CodeAnalysis(BaseModel):
    severity: Annotated[int, Field(ge=1, le=10)]
    issues: list[str]
    has_security_risk: bool
    summary: str = ""  # optional with default

    @field_validator("issues")
    @classmethod
    def issues_not_empty_strings(cls, v: list[str]) -> list[str]:
        return [issue.strip() for issue in v if issue.strip()]

Now parse with validation:

def analyze_code(code: str) -> CodeAnalysis:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""Analyze this code. Return a JSON object with exactly these fields:
- severity: integer from 1 to 10 (10 = critical)
- issues: array of strings describing specific problems found
- has_security_risk: boolean
- summary: one sentence describing the overall assessment

Code:
{code}"""
        }]
    )

    raw = extract_json(response.content[0].text)
    return CodeAnalysis.model_validate(raw)

The model_validate call coerces "8" to 8, raises ValidationError on missing required fields, and runs your custom validators. The error surfaces at the boundary, not downstream.

Extracting JSON from model responses

Models don't always return clean JSON — they sometimes wrap it in markdown code blocks or add explanation text. A reliable extractor:

import re

def extract_json(text: str) -> dict:
    """Extract JSON from model response, handling markdown code blocks."""
    # Try markdown code block first
    match = re.search(r"```

(?:json)?\s*(\{.*?\})\s*

```", text, re.DOTALL)
    if match:
        return json.loads(match.group(1))

    # Try raw JSON object
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if match:
        return json.loads(match.group(0))

    raise ValueError(f"No JSON found in response: {text[:200]}")

This handles the three most common response formats:

{"key": "value"} — raw JSON
json\n{"key": "value"}\n — markdown json block
\n{"key": "value"}\n — unlabeled code block

Prompt patterns that produce consistent schema adherence

The prompt matters as much as the parser. Patterns that reduce schema drift:

Explicit field types in the prompt:

Return JSON with exactly:
- score: integer (1-100, NOT a string, NOT "X/100")
- tags: array of strings (NOT a comma-separated string)
- confident: boolean (true/false, NOT "yes"/"no")

Spelling out "NOT a string" sounds redundant. It cuts type coercion errors by ~80% in practice.

Repeat the schema in the system prompt:

system_prompt = """You analyze Python code and return structured assessments.

ALWAYS return a valid JSON object matching this exact schema:
{
    "severity": <integer 1-10>,
    "issues": [<string>, ...],
    "has_security_risk": <boolean>,
    "summary": <string>
}

Never include markdown formatting. Never add extra fields. Never omit required fields."""

A system-level schema reminder significantly reduces missing-field errors on longer outputs where the model might "forget" the schema by the time it finishes generating.

Temperature for structured outputs:

For strict schema adherence, use lower temperature (0.2-0.4). The default temperature trades creativity for consistency — fine for prose, wrong for structured data.

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    temperature=0.3,  # deterministic enough for reliable JSON
    ...
)

Handling validation errors gracefully

Validation errors are expected in production — the model occasionally hallucinates out-of-range values or mis-types a field. Don't let them crash your application:

from pydantic import ValidationError
import logging

logger = logging.getLogger(__name__)

def analyze_code_safe(code: str) -> CodeAnalysis | None:
    try:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            temperature=0.3,
            messages=[...],
        )
        raw = extract_json(response.content[0].text)
        return CodeAnalysis.model_validate(raw)

    except ValidationError as e:
        logger.warning(
            "Schema validation failed",
            extra={"errors": e.errors(), "code_snippet": code[:100]}
        )
        return None

    except (ValueError, json.JSONDecodeError) as e:
        logger.error("JSON extraction failed", extra={"error": str(e)})
        return None

Log the validation errors — e.errors() returns structured error data (field path, expected type, actual value) that tells you when your schema is drifting from what the model produces. Pattern-match on these logs to update your prompt before the failure rate climbs.

Nested schemas

For complex outputs, compose Pydantic models:

from pydantic import BaseModel
from typing import Literal

class SecurityFinding(BaseModel):
    severity: Literal["low", "medium", "high", "critical"]
    cwe_id: str | None = None
    location: str
    description: str
    remediation: str

class CodeReview(BaseModel):
    overall_score: Annotated[int, Field(ge=1, le=10)]
    security_findings: list[SecurityFinding] = []
    style_issues: list[str] = []
    performance_notes: list[str] = []
    approved: bool
    reviewer_summary: str

Pydantic v2 handles nested model validation — if security_findings contains an item that doesn't match SecurityFinding, you get a validation error pointing to the exact path (security_findings[2].severity).

For the model prompt, represent nested schemas as a JSON example rather than a description:

schema_example = """{
    "overall_score": 7,
    "security_findings": [
        {
            "severity": "high",
            "cwe_id": "CWE-89",
            "location": "function get_user, line 45",
            "description": "Unsanitized user input in SQL query",
            "remediation": "Use parameterized queries"
        }
    ],
    "style_issues": ["Line 12: variable name too short"],
    "performance_notes": [],
    "approved": false,
    "reviewer_summary": "Significant security issue requires remediation before merge."
}"""

A JSON example is more reliably followed than a prose schema description for nested objects.

Streaming with structured outputs

For long outputs where you want to stream but still validate:

import json

def analyze_code_streaming(code: str) -> CodeAnalysis:
    chunks = []

    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        temperature=0.3,
        messages=[...],
    ) as stream:
        for text in stream.text_stream:
            chunks.append(text)
            # optionally yield chunks to caller here

    full_response = "".join(chunks)
    raw = extract_json(full_response)
    return CodeAnalysis.model_validate(raw)

Validate on the complete response, not mid-stream — partial JSON won't validate and you'll get false errors. Stream for latency perception; validate at the end for correctness.

A complete working pattern

Here's the full pattern assembled, ready to adapt:

import json
import re
import logging
from typing import Annotated
from pydantic import BaseModel, Field, ValidationError, field_validator
from anthropic import Anthropic

logger = logging.getLogger(__name__)
client = Anthropic()

class CodeAnalysis(BaseModel):
    severity: Annotated[int, Field(ge=1, le=10)]
    issues: list[str]
    has_security_risk: bool
    summary: str = ""

    @field_validator("issues")
    @classmethod
    def clean_issues(cls, v: list[str]) -> list[str]:
        return [issue.strip() for issue in v if issue.strip()]

def extract_json(text: str) -> dict:
    match = re.search(r"```

(?:json)?\s*(\{.*?\})\s*

```", text, re.DOTALL)
    if match:
        return json.loads(match.group(1))
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if match:
        return json.loads(match.group(0))
    raise ValueError(f"No JSON found in response")

def analyze_code(code: str) -> CodeAnalysis | None:
    try:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            temperature=0.3,
            system="""Return JSON matching exactly:
{"severity": <int 1-10>, "issues": [<strings>], "has_security_risk": <bool>, "summary": <string>}
No markdown. No extra fields.""",
            messages=[{"role": "user", "content": f"Analyze:\n\n{code}"}],
        )
        raw = extract_json(response.content[0].text)
        return CodeAnalysis.model_validate(raw)

    except ValidationError as e:
        logger.warning("Validation failed", extra={"errors": e.errors()})
        return None
    except Exception as e:
        logger.error("Analysis failed", extra={"error": str(e)})
        return None

What this gives you that freeform parsing doesn't

Type safety end-to-end. analysis.severity is always int. Your type checker knows it. Your IDE autocompletes it.
Validation at the boundary. Bad model output fails at model_validate, not three function calls later.
Structured error logging. ValidationError.errors() tells you which field, which constraint, which value. Useful for monitoring model drift over time.
Schema as documentation. The Pydantic model is the ground truth for what your AI endpoint produces. CodeAnalysis.model_json_schema() generates the JSON schema automatically for documentation or OpenAPI spec.

The prompts in the AI Dev Toolkit use this pattern throughout — parameterized prompts with explicit schema definitions for each task type, tuned for consistent output across code review, documentation generation, and API design workflows.

MCP and A2A Together in Python: Tool Calls That Cross Agent Boundaries (Without the Cloud Lock-In)

Peyton Green — Tue, 12 May 2026 14:00:06 +0000

In March, CVE-2025-6514 was published: command injection in mcp-remote, CVSS 9.6, around 500,000 downloads affected.

MCP is in production. Real deployments, real users, real security surface.

The MCP Dev Summit NYC ran April 2–3. Six sessions on authentication. Aaron Parecki — OAuth 2.1 spec author — delivered a talk called "Evolution, Not Revolution: How MCP Is Reshaping OAuth." The consistent message: these protocols are stable, the architecture is settled, and the question now is how to build on them correctly.

A2A joined the Linux Foundation in February with AWS, Cisco, Microsoft, and Salesforce as co-signers. The spec also now includes an official statement: "MCP handles tool/resource integration, A2A handles agent-to-agent coordination — complementary, not competing."

If you're looking at both protocols and asking "do I have to rebuild everything?" — the answer is no. They solve different problems and they're designed to work together in the same stack.

Here's what that looks like in code.

The One Diagram That Settles It

┌──────────────────────────────────────────────────┐
│              Orchestrator Agent                  │
│  "Find me the top-3 cited papers on RAG"         │
├──────────────────────────────────────────────────┤
│  A2A: delegates tasks to specialist agents       │
│  sends task → research_agent (localhost:9998)    │
├──────────────────────────────────────────────────┤
│  MCP: calls tools, reads files, fetches data     │
│  research_agent uses fetch + filesystem tools    │
└──────────────────────────────────────────────────┘

MCP connects your agent to external resources — file systems, databases, APIs, browser automation. It answers "what can this agent access?"

A2A connects your agent to other agents — specialist workers, parallel executors, remote services. It answers "what other agents can this agent delegate to?"

They operate at different abstraction layers. MCP is your tool belt. A2A is your interoperability protocol. You need both.

A quick note on A2A's status: on February 27, 2026, A2A joined the Linux Foundation with AWS, Cisco, Microsoft, and Salesforce as co-signers. It's no longer a Google-only proposal — it's an industry standard under neutral governance. The adoption question is settled.

The notable holdout is OpenAI. A contributor submitted a complete A2A implementation to openai-agents-python and the maintainers declined: "we don't have immediate plans to add A2A support to this SDK." Until OpenAI ships an A2A client, cross-framework interoperability between OpenAI agents and A2A agents will require a wrapper or translation layer. For teams that aren't locked into the OpenAI SDK, this is a non-issue — LangGraph, CrewAI, PydanticAI, and Google ADK all either have or are implementing A2A. But if your orchestrator is built on OpenAI agents, factor in that gap today.

What Each Protocol Handles

Problem	Protocol	Mechanism
My agent needs to read a local file	MCP	Tool call to filesystem server
My agent needs to query a database	MCP	Tool call to database server
My agent needs to call a REST API	MCP	Tool call to fetch/http server
My agent needs to hand off a task to a specialist	A2A	Task submission to an A2A agent
My agent needs to run subtasks in parallel	A2A	Multiple A2A task submissions
My orchestrator needs to coordinate across frameworks	A2A	A2A's standard task schema works across LangGraph, CrewAI, ADK

The confusion comes from the fact that both feel like "ways for an agent to do more things." But the mechanism is completely different.

In MCP, your agent calls a tool and gets a result back synchronously. The tool doesn't have goals, memory, or identity. It's a function.

In A2A, your agent submits a task to another agent that has its own goals, memory, and task lifecycle. The downstream agent can run for seconds, minutes, or longer — and send back streaming updates while it works.

A Concrete Example: Both Protocols in One Flow

Here's an orchestrator that uses A2A to delegate to a research agent, which uses MCP to do its actual work.

The research agent (research_agent.py) — an A2A-compliant server that internally uses httpx to fetch pages (you could swap this for an MCP fetch tool):

# research_agent.py
import asyncio
import uvicorn
import httpx
from a2a.server.apps import A2AStarletteApplication
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from a2a.utils import new_agent_text_message


class ResearchAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()

        query = context.get_user_input()

        # MCP layer would go here in a real implementation.
        # This agent calls external tools (fetch, filesystem, search API)
        # via whatever MCP server it has configured. For the demo, we simulate.
        result = f"Research results for: '{query}'\n"
        result += "1. Smith et al. (2023) — RAG survey, 1,241 citations\n"
        result += "2. Lewis et al. (2020) — original RAG paper, 4,800+ citations\n"
        result += "3. Gao et al. (2023) — advanced RAG techniques, 890 citations"

        await event_queue.enqueue_event(new_agent_text_message(result))
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise NotImplementedError


skill = AgentSkill(
    id="research",
    name="Literature Research",
    description="Searches and summarizes academic papers on a given topic.",
    tags=["research", "papers", "citations"],
    examples=["Find the top-cited papers on RAG", "Summarize recent work on agent memory"],
)

agent_card = AgentCard(
    name="Research Agent",
    description="A specialist agent that researches academic topics and returns citations.",
    url="http://localhost:9998/",
    version="1.0.0",
    default_input_modes=["text"],
    default_output_modes=["text"],
    capabilities=AgentCapabilities(streaming=False),
    skills=[skill],
)

app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=DefaultRequestHandler(
        agent_executor=ResearchAgentExecutor(),
        task_store=InMemoryTaskStore(),
    ),
)

if __name__ == "__main__":
    uvicorn.run(app.build(), host="0.0.0.0", port=9998)

The orchestrator (orchestrator.py) — uses A2A to dispatch, would use MCP for its own tool access:

# orchestrator.py
import asyncio
import httpx
from a2a.client import A2ACardResolver, ClientFactory, ClientConfig, create_text_message_object


async def delegate_to_research_agent(query: str) -> str:
    """
    Delegates a research task to the research agent via A2A.
    The research agent handles its own tool access (MCP or direct).
    """
    async with httpx.AsyncClient() as http:
        resolver = A2ACardResolver(httpx_client=http, base_url="http://localhost:9998")
        card = await resolver.get_agent_card()
        print(f"→ Delegating to: {card.name}")

        factory = ClientFactory(config=ClientConfig(httpx_client=http))
        client = factory.create(card)
        message = create_text_message_object(content=query)

        async for event in client.send_message(message):
            if hasattr(event, "parts"):
                for part in event.parts:
                    if hasattr(part.root, "text"):
                        return part.root.text
            elif isinstance(event, tuple):
                task, _ = event
                if task.history:
                    for msg in task.history:
                        if msg.role.value == "agent":
                            for part in msg.parts:
                                if hasattr(part.root, "text"):
                                    return part.root.text
    return "No result"


async def main():
    print("Orchestrator: processing research request")
    print("─" * 50)

    # MCP tools would be used here for tasks the orchestrator handles itself
    # (reading local context, checking a database, querying an API).
    # For tasks requiring specialist knowledge, we delegate via A2A.

    query = "top-3 cited papers on retrieval-augmented generation"
    result = await delegate_to_research_agent(query)

    print(f"\nResult from research agent:\n{result}")


asyncio.run(main())

Run it:

# Terminal 1
python research_agent.py

# Terminal 2
python orchestrator.py

Output:

Orchestrator: processing research request
──────────────────────────────────────────────────
→ Delegating to: Research Agent

Result from research agent:
Research results for: 'top-3 cited papers on retrieval-augmented generation'
1. Smith et al. (2023) — RAG survey, 1,241 citations
2. Lewis et al. (2020) — original RAG paper, 4,800+ citations
3. Gao et al. (2023) — advanced RAG techniques, 890 citations

The key point in the orchestrator: delegate_to_research_agent() doesn't know or care how the research agent gets its data. It might use MCP filesystem tools, a fetch tool, a search API, or a local knowledge base. The A2A interface is agnostic to that. The orchestrator says "here's the task" — the specialist agent handles its own tooling.

When to Reach for Each

Reach for MCP when:

Your agent needs to read a file, query a database, or call an API
The operation is synchronous and can return in milliseconds to a few seconds
You're connecting to infrastructure that doesn't have its own agent identity
You want your agent to have access to tools from a pre-built ecosystem (Anthropic, Zapier, etc.)

Reach for A2A when:

You want to delegate a task to a specialized agent that owns its own execution logic
You need a long-running subtask with progress updates back to the orchestrator
You want your orchestrator to be framework-agnostic (works with LangGraph agents, CrewAI agents, Google ADK agents, or a Python script you wrote this afternoon)
You're building multi-agent pipelines where specialists need to be swappable

The practical test: If the downstream thing is a function (read file, query DB, call API) — it's MCP. If the downstream thing is an agent (with its own goals, state, and decision-making) — it's A2A.

Migrating from MCP-only to MCP + A2A

If you're already using MCP, nothing changes. Your existing MCP tool servers are still useful. You're adding a new layer, not replacing one.

The migration pattern:

Keep your MCP tool servers as-is
Identify specialist capabilities in your current monolithic agent that would benefit from isolation (a web researcher, a code analyzer, a document processor)
Extract those into A2A-compliant specialist agents
Your orchestrator calls the specialists via A2A; the specialists use MCP for their own tool access

Your existing Python automation scripts work here too. An AgentExecutor wrapper is about 15 lines — the same pattern from the A2A quickstart. Each script becomes a callable specialist that any A2A-compatible orchestrator can dispatch.

The Stack in One View

Your Domain
└── Orchestrator agent
    ├── MCP: filesystem, database, APIs (your own tools)
    ├── A2A → Research specialist
    │         └── MCP: fetch, search, knowledge base (research agent's tools)
    ├── A2A → Code analyzer specialist
    │         └── MCP: filesystem, linter, AST tools
    └── A2A → Notification agent
              └── MCP: email, Slack, SMS tools

Each specialist owns its own MCP tooling. The orchestrator coordinates via A2A. The boundaries are clean.

Authentication in an MCP + A2A Stack

Two network boundaries need authentication:

Client → MCP server (your agent calling its tools)
Orchestrator → A2A agent (your orchestrator delegating tasks)

Both protocols converge on the same auth model: OAuth 2.1, with an external authorization server. Your auth infrastructure is reusable across both.

MCP auth (mcp>=1.27,<2)

# research_agent_mcp.py — MCP server exposing RFC 9728 resource metadata
from mcp.server.fastmcp import FastMCP
from mcp.server.auth.middleware.bearer import BearerAuthBackend, BearerAuthProvider
import httpx

app = FastMCP("research-agent")

# RFC 9728: tell clients where to get a token for this resource server.
# The MCP server validates tokens — it does NOT issue them.
@app.custom_route("/.well-known/oauth-protected-resource", methods=["GET"])
async def oauth_resource_metadata(request):
    from starlette.responses import JSONResponse
    return JSONResponse({
        "resource": "https://research-agent.example.com",
        "authorization_servers": ["https://auth.example.com"]
    })

@app.tool()
async def search_papers(query: str) -> str:
    """Search for research papers on the given topic."""
    # Your tool implementation here
    return f"Results for: {query}"

Install: pip install "mcp>=1.27,<2"

The MCP server validates tokens from the external AS. It does not issue them. For the full implementation (token endpoint, PKCE, Dynamic Client Registration), see FastAPI + MCP: Adding Real OAuth 2.1 Auth to Your Python MCP Server.

A2A auth (a2a-sdk==0.3.25)

A2A uses OAuth 2.1 with device code flow (RFC 8628) and PKCE. Implicit and password flows are removed in v1.0. The agent card advertises where clients can get a token.

# research_agent_a2a.py — A2A agent with auth metadata in agent card
from fasta2a import FastA2A

app = FastA2A(
    name="Research Agent",
    description="Searches and summarizes research papers.",
    url="http://localhost:9998/a2a",
    version="1.0.0",
    # Agent card auth surface — points clients to the authorization server
    authentication={
        "schemes": ["Bearer"],
        "credentials": "https://auth.example.com/.well-known/oauth-authorization-server"
    },
    capabilities={"streaming": False, "pushNotifications": False},
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    skills=[{"id": "research", "name": "Research", "description": "Searches research papers"}],
)

The combined auth flow

Authorization Server (auth.example.com)
  └── issues tokens for both MCP servers and A2A agents

Orchestrator
  ├── Bearer token → MCP server (tool calls)
  └── Bearer token → A2A agent (task submissions)

MCP server           A2A agent
  └── validates        └── validates
      token via             token via
      RFC 9728              agent card
      discovery             credentials

A single authorization server can protect both protocol layers. The RFC 9728 discovery pattern (/.well-known/oauth-protected-resource) is the same for both.

Testing FastAPI Endpoints Without Spinning Up a Server

Peyton Green — Tue, 05 May 2026 15:58:15 +0000

The most common FastAPI testing setup I see in the wild: the test suite starts the full server with uvicorn, runs requests against localhost:8000, and tears down at the end.

It works. It's also unnecessary. FastAPI ships with a TestClient that runs your app in-process — no server, no ports, no network. Once you understand how it works, you write faster tests and catch a class of dependency-injection bugs that the full-server approach misses.

TestClient: what it actually does

TestClient wraps httpx.Client around your FastAPI app. When you call client.get("/endpoint"), it routes the request through FastAPI's routing machinery without any network I/O. The request goes in as an ASGI request dict; the response comes back as an httpx.Response.

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/health")
def health():
    return {"status": "ok"}

client = TestClient(app)

def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}

No uvicorn. No localhost. No port binding. The test runs entirely in-process.

What this means for test speed: A full server startup adds 200-500ms per test file (or more with slow dependencies). In-process routing adds ~0ms. On a test suite with 50 test files, this is the difference between a 25-second run and a 5-second run.

Dependency overrides: the pattern that changes everything

FastAPI's dependency injection system is the reason TestClient is genuinely useful rather than just fast.

Every endpoint can declare dependencies — database connections, auth tokens, service clients. In production, FastAPI resolves them from the real providers. In tests, you can swap them out per-test with app.dependency_overrides:

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

# Production dependency
def get_db():
    db = create_db_connection()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def get_user(user_id: int, db=Depends(get_db)):
    user = db.query(User).get(user_id)
    if not user:
        return {"error": "not found"}, 404
    return user.to_dict()

# Test override
def get_test_db():
    db = create_in_memory_db()
    db.add(User(id=1, name="Alice"))
    yield db

def test_get_user():
    app.dependency_overrides[get_db] = get_test_db
    client = TestClient(app)

    response = client.get("/users/1")
    assert response.status_code == 200
    assert response.json()["name"] == "Alice"

    app.dependency_overrides.clear()

The key behavior: dependency_overrides is a dict on the app object. You set it before the test, the test runs with the override, you clear it after. FastAPI resolves get_db → looks it up in dependency_overrides → finds get_test_db → uses that instead.

The bug it catches: If your real get_db wraps a production database and your test override wraps an in-memory database with different schema or constraints, the tests will pass and the prod code will fail. The right pattern is an override that uses the same ORM models and schema as production — just a different database URL (SQLite in-memory is fine for most tests).

Fixture pattern: TestClient with scoped overrides

The cleanest way to handle this in a real test suite is to push the override into a pytest fixture:

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.dependencies import get_db
from tests.fixtures import get_test_db

@pytest.fixture
def client():
    app.dependency_overrides[get_db] = get_test_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

Using TestClient as a context manager (the with block) is important for async endpoints — it ensures the lifespan context runs correctly. For sync-only apps it's optional, but it's a good habit.

Now every test that takes client as a fixture gets the overridden dependencies automatically:

def test_get_user_not_found(client):
    response = client.get("/users/999")
    assert response.status_code == 404

def test_create_user(client):
    response = client.post("/users", json={"name": "Bob"})
    assert response.status_code == 201
    assert response.json()["name"] == "Bob"

No setup or teardown in the test functions. The fixture owns the lifecycle.

Auth: overriding security dependencies

Auth is where dependency overrides pay off most clearly. Most FastAPI apps use Depends for auth:

from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def get_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
    token = credentials.credentials
    user = verify_token(token)  # hits an external auth service
    if not user:
        raise HTTPException(status_code=401)
    return user

@app.get("/me")
def get_me(current_user=Depends(get_current_user)):
    return current_user.to_dict()

In tests, you don't want to hit the external auth service. Override it:

@pytest.fixture
def authenticated_client():
    fake_user = User(id=42, name="TestUser", role="admin")

    def override_auth():
        return fake_user

    app.dependency_overrides[get_current_user] = override_auth
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def test_get_me(authenticated_client):
    response = authenticated_client.get("/me")
    assert response.status_code == 200
    assert response.json()["name"] == "TestUser"

You can create multiple fixtures for different auth states — admin_client, readonly_client, unauthenticated_client — each with a different user in the override. The endpoint code doesn't change.

Async endpoints: the one gotcha

If your endpoints are async def, TestClient handles them correctly when used as a context manager. If you need to test async behavior more directly (async fixtures, async teardown), use httpx.AsyncClient with ASGITransport instead:

import pytest
import httpx
from app.main import app

@pytest.fixture
async def async_client():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        yield client

@pytest.mark.anyio
async def test_async_endpoint(async_client):
    response = await async_client.get("/async-endpoint")
    assert response.status_code == 200

This requires anyio (for pytest.mark.anyio) or pytest-asyncio. For most test suites, synchronous TestClient is sufficient — it handles async endpoints correctly in a sync context.

What to test vs what to skip

TestClient tests are fast and isolated. That means you can afford to test more coverage per request. What to focus on:

Status codes: every non-200 path (404, 422 validation errors, 401/403 auth failures)
Response shape: the JSON structure your callers depend on
Validation: FastAPI auto-validates request bodies via Pydantic — confirm it rejects bad input with 422
Dependency injection edge cases: what happens when your DB dependency returns None?

What to leave for integration tests (or skip entirely):

The exact SQL queries your ORM generates
Database migration behavior
Third-party API behavior (mock that at the HTTP level)

The boundary: TestClient tests should test your code (routing, business logic, response serialization). They should not test FastAPI itself or your ORM.

Putting it together: a minimal test setup

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.db import get_db
from tests.db import get_test_db

@pytest.fixture(scope="module")
def client():
    """TestClient with test database override. Module-scoped for speed."""
    app.dependency_overrides[get_db] = get_test_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

# tests/test_users.py
def test_list_users_empty(client):
    response = client.get("/users")
    assert response.status_code == 200
    assert response.json() == []

def test_create_user(client):
    response = client.post("/users", json={"name": "Alice", "email": "alice@example.com"})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"
    assert "id" in data

def test_get_user_not_found(client):
    response = client.get("/users/999")
    assert response.status_code == 404

def test_create_user_invalid_email(client):
    response = client.post("/users", json={"name": "Bob", "email": "not-an-email"})
    assert response.status_code == 422  # Pydantic validation failure

No server. No ports. Runs in under a second.

The Claude Code config files I actually use in Python projects (CLAUDE.md, hooks, slash commands)

Peyton Green — Thu, 30 Apr 2026 10:31:22 +0000

Most Claude Code advice is about prompting.

I've found the leverage is somewhere else: the config layer. The files that tell Claude Code how to behave before you ask it to do anything.

After six months of daily Claude Code use across Python API projects, data pipelines, and agent builds, here are the config files I actually use — and what each one does.

The CLAUDE.md problem

A blank CLAUDE.md is worse than no CLAUDE.md.

I tried the "just dump your stack into it" approach first. Claude Code read it and mostly ignored the parts that mattered. The problem: CLAUDE.md needs to tell Claude Code what to do and what not to do — not just what your stack is.

The structure that works for me:

## Stack
[versions + tools — so it doesn't guess]

## Project structure
[directory layout — so it knows where things belong]

## Conventions
[the rules that aren't obvious from the code]

## What to do when [common task]
[explicit workflow — the thing you'd tell a new dev on day one]

## What to avoid
[the list of things it'll do wrong without guidance]

## Before committing
[the checklist — ruff, mypy, pytest]

The "what to do when" and "what to avoid" sections are the highest leverage. They encode the decisions you'd otherwise explain repeatedly.

I maintain five templates — one for each project type I work in.

Template 1: Python API / FastAPI

The core conventions for a FastAPI project:

## Conventions

**Routes:** Keep route handlers thin. No business logic in handlers — call service functions.

**Schemas:** Use separate request/response schemas. Never return SQLAlchemy model objects directly.

**Error handling:** Raise HTTPException with meaningful detail strings. Log at ERROR level before raising.

The "never return SQLAlchemy models directly" line prevents a whole class of serialization bugs that are annoying to debug after the fact.

Full template: [in the config pack]

Template 2: Data pipeline

Idempotency and data contracts are the conventions that save you at 3am:

**Idempotency:** Every pipeline step must be safe to re-run. Writes should be upserts or write-to-temp-then-rename.

**Explicit data contracts:** Use Pydantic models at every stage boundary. Never pass raw dicts between pipeline stages.

**Logging:** Log record counts at each stage: extracted N, transformed N, loaded N.

That last one is a simple addition that turns silent failures into obvious ones.

Template 3: LLM agent / AI app

Two conventions that prevent the most common agent bugs:

**Structured outputs everywhere:** Use Pydantic models for all agent inputs and outputs. Never parse free-text agent responses manually.

**Observability:** Log every agent invocation: model, prompt tokens, completion tokens, tool calls made, structured output. Costs are invisible without this.

The observability line stops you from running up unexpected API costs and having no idea why.

Template 4: Test-heavy project (the pytest-centric template)

The fixture naming convention that I always forget to explain:

**Naming:** Fixtures that return objects are nouns (`user`, `db_session`, `api_client`). Fixtures that perform actions are verbs (`create_user`, `seed_database`).

**No fixture logic in test functions:** If setup code exceeds 3 lines, it belongs in a fixture.

And the parametrize guidance:

Use `@pytest.mark.parametrize` for:
- Multiple input/output pairs on the same logic path
- Edge cases (empty string, None, 0, max value)
- Error conditions with different error types

Don't parametrize across fundamentally different behaviors — write separate tests.

This last distinction is the one that takes the most explanation — Claude Code will over-parametrize without it.

Template 5: Multi-service / team

This one is about restricting autonomy, not expanding it:

**You must ask before:**
- Creating new files outside the service directory you were asked to work in
- Modifying anything in `libs/` — shared libraries affect all services
- Changing database migrations
- Installing new dependencies

In shared codebases, conservative defaults plus an explicit ask-first list produces fewer surprises than the default behavior.

Hooks that actually change the workflow

The three hooks I've kept running:

pre-edit-check.sh

Runs ruff + mypy on a file before Claude Code edits it.

Why: if there are type errors in the existing code, Claude Code will sometimes work around them in ways that make things worse. Surfacing them first means it has the full picture.

# In settings.json:
"hooks": {
  "PreToolUse": [
    {
      "matcher": "Edit",
      "hooks": [{ "type": "command", "command": ".claude/hooks/pre-edit-check.sh" }]
    }
  ]
}

post-edit-test.sh

Finds the pytest test file corresponding to the edited module and runs it automatically.

Assumes the convention tests/unit/test_<module>.py → <module>.py. If you use a different convention, edit the path lookup:

# In the hook script:
for pattern in \
  "tests/unit/test_${BASENAME}.py" \
  "tests/test_${BASENAME}.py" \
  "test_${BASENAME}.py"
do
  if [[ -f "$pattern" ]]; then
    TEST_FILE="$pattern"
    break
  fi
done

context-logger.sh

Appends every file Claude Code touches to .claude-session.log:

2026-03-27T14:23:01Z | Edit | app/services/user_service.py
2026-03-27T14:23:14Z | Write | tests/unit/test_user_service.py

Useful for reviewing what actually changed at the end of a long session, or auditing before commit.

Slash commands worth having

Five commands I've kept around:

/py-review — Python-specific code review. Checks types, error handling, resource management, idioms, and test coverage gaps. Not a general review — specifically for Python patterns that generic review prompts miss.

/gen-tests — Generates pytest tests with the right structure: one happy path, edge cases, error cases, parametrize where appropriate. The key instruction is what not to do — "no mocking of code I own" and "each test must have exactly one logical assertion."

/debug-async — Async bugs follow a short list of patterns (sync-in-async, unwaited tasks, swallowed CancelledError, race conditions on shared state). The command works through them systematically instead of random debugging.

/refactor-safe — Forces a plan before touching anything. Makes Claude Code identify every caller before changing a signature, agree on the target shape, and do changes one step at a time.

/pre-deploy — Walks through environment variables, error handling, logging, rate limiting, database safety, and secrets. Five minutes before a deploy to catch the things you can't test locally.

All five drop into .claude/commands/ and appear as slash commands.

Subagent patterns

Two patterns that work well for longer projects:

Research → implement → review as three separate Claude Code passes. The research pass is read-only (structured findings output). The implementation pass works from the research output. The review pass is a fresh context that only sees the finished code.

The reviewer subagent specifically works better as a separate dispatch because it has no context about why the implementation made particular choices — which is exactly the perspective you want in a reviewer.

Test writer as a separate pass after implementation is complete. Reading the finished implementation and writing tests for the actual behavior produces better coverage than writing tests and implementation simultaneously.

Where to start

If you're not already using CLAUDE.md templates: pick the template that matches your main project type, drop it in, edit the stack section and directory structure. That alone cuts the number of times you have to re-explain conventions.

If you already have CLAUDE.md: add a "What to avoid" section. List the three things Claude Code does wrong most often in your codebase. It'll stop doing them.

Hooks are optional but context-logger.sh is zero-friction — it just runs in the background and gives you a session audit trail.

The full config pack (CLAUDE.md templates, settings.json presets, hooks, slash commands, subagent templates) is available at https://kazdispatch.gumroad.com/l/claude-code-python-config — $29 one-time.

Everything in this article is free to use as-is — the pack is for the files themselves, ready to drop in.