DEV Community: Peyton Green

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.

Python Type Hints That Actually Catch Bugs (Not Just Satisfy mypy)

Peyton Green — Tue, 28 Apr 2026 10:13:01 +0000

Most type hint guides teach you syntax. This one is about which annotations actually prevent bugs.

After two years of adding types to codebases that didn't have them — and watching which annotations caught real issues and which were just ceremony — I've settled on four patterns. They're not the four most commonly taught. They're the four that have caught production bugs before they shipped.

The gap between "mypy passes" and "your code is correct"

mypy can pass on code that will throw a TypeError at runtime. This happens often enough that it's worth understanding why.

The core issue: mypy checks the annotations you write, but it can't verify the annotations are accurate. If you annotate a function as accepting str when it sometimes receives None, mypy will accept the annotation and check callers against it — but the function will still fail at runtime when None shows up.

This matters because the value of type hints isn't in satisfying the type checker. It's in forcing yourself to be precise about what your code actually accepts and returns. The annotation is a claim. The bug comes from making a wrong claim.

The patterns below are the ones where getting the annotation right catches a class of bug at annotation time rather than runtime.

Pattern 1: `Optional[X]` vs `Union[X, None]` — and why you should use neither

Optional[str] is syntactic sugar for Union[str, None]. It's the most common type annotation that developers get wrong in a consistent direction: they use it too rarely when they mean "this can be None" and too often when they mean "this is usually a string but I'm not sure."

The bug pattern it catches:

# Without types
def get_user_name(user_id: int):
    user = db.get(user_id)
    return user.name  # AttributeError if user is None

# With the wrong annotation — mypy doesn't help you
def get_user_name(user_id: int) -> str:
    user = db.get(user_id)
    return user.name  # mypy is happy; still crashes at runtime

# With the right annotation — mypy catches the unguarded access
def get_user_name(user_id: int) -> Optional[str]:
    user = db.get(user_id)
    if user is None:
        return None
    return user.name  # safe

The rule: use Optional[X] only when you've verified that the value actually can be None and you're prepared to handle it at every call site. The annotation should reflect reality, not intent.

Modern Python (3.10+) allows str | None as equivalent syntax. Use whichever your codebase has standardized on.

What it catches: Unguarded None access. The annotation forces you to decide whether None is a valid state, and mypy will flag every call site that treats the return value as definitely non-None without a guard.

Pattern 2: `TypeGuard` for narrowing in conditionals

Type narrowing is how mypy (and your IDE) understand what type a variable has inside a conditional block. The built-in narrowing handles the common cases:

def process(value: str | int) -> str:
    if isinstance(value, str):
        return value.upper()  # mypy knows value is str here
    return str(value)         # mypy knows value is int here

But narrowing fails when the check is in a helper function:

def is_string(value: object) -> bool:
    return isinstance(value, str)

def process(value: str | int) -> str:
    if is_string(value):
        return value.upper()  # mypy error: int has no attribute upper
    return str(value)

mypy can't infer that is_string narrows the type. TypeGuard fixes this:

from typing import TypeGuard

def is_string(value: object) -> TypeGuard[str]:
    return isinstance(value, str)

def process(value: str | int) -> str:
    if is_string(value):
        return value.upper()  # mypy is now happy — and correct
    return str(value)

What it catches: Type errors that only appear when narrowing is delegated to a helper. This pattern comes up frequently in validation code — the kind of code that checks "is this a valid X?" before dispatching on it. Without TypeGuard, your validation code provides no type safety downstream.

Pattern 3: `Protocol` for duck typing that mypy understands

Python is built on duck typing — if it has the right methods, it works. Type annotations fight this by default, because isinstance checks against concrete classes.

Protocol bridges the gap:

from typing import Protocol

class Serializable(Protocol):
    def to_dict(self) -> dict: ...

def save(item: Serializable) -> None:
    data = item.to_dict()
    # persist data...

Now any class with a to_dict() -> dict method satisfies Serializable — no inheritance required. mypy checks structural compatibility at the call site.

The bug pattern this catches is subtle. Without Protocol, you have two options for typing heterogeneous inputs:

Use Any — no type safety
Create a common base class — requires touching all implementations, breaks if they're in third-party code

With Protocol, you can annotate exactly the interface you use. This matters most for:

Library boundaries — you want to accept anything that looks like a file, a logger, or a cache, regardless of concrete type
Test doubles — your mock objects can satisfy the protocol without inheriting from the real class
Third-party classes — you can write a protocol that matches a third-party class without modifying it

class HasClose(Protocol):
    def close(self) -> None: ...

def cleanup(resource: HasClose) -> None:
    resource.close()

# Works with anything that has close(): file objects, database connections,
# custom resources, test doubles — no inheritance required

What it catches: Type errors at the boundary between components. If your component uses three methods of an object, Protocol lets you express exactly that dependency. If a caller passes something that doesn't have those three methods, mypy catches it.

Pattern 4: `overload` for functions with multiple valid signatures

Some functions behave differently depending on argument type. The usual workaround is Union return types, but that forces callers to handle cases that can't actually occur:

def parse(value: str | bytes) -> str | bytes:
    if isinstance(value, bytes):
        return value.decode()
    return value

result = parse("hello")
result.upper()  # mypy error: bytes has no attribute upper
# mypy doesn't know that str input → str output

@overload solves this:

from typing import overload

@overload
def parse(value: str) -> str: ...
@overload
def parse(value: bytes) -> str: ...

def parse(value: str | bytes) -> str:
    if isinstance(value, bytes):
        return value.decode()
    return value

result = parse("hello")
result.upper()  # correct — mypy knows this is str

The overloaded signatures are the type-checker declarations; the actual implementation is below them. mypy uses the overloads to determine the return type at each call site.

What it catches: Return type ambiguity in multi-dispatch functions. This is common in any function that accepts multiple input types and transforms them — parsers, converters, and coercions. Without overload, you either use Any (no safety) or accept false errors at call sites (noise that causes engineers to disable type checking).

When type hints don't help

Type hints are least valuable when:

You're already handling all the cases (the annotation confirms what you already know)
The function is trivially short (the annotation adds ceremony without reducing cognitive load)
You're annotating purely for documentation rather than checking (mypy ignores annotations on unannotated call sites)

They're most valuable at boundaries — function signatures that cross module or component lines, public API surfaces, places where data changes shape. The interior of a private implementation function is often better left unannotated; the signature that other modules call should almost always be.

A practical approach: run mypy --strict on your public API surface and your test fixtures. Let the interior be more permissive. You'll catch the bugs that matter without spending time on annotations that don't.

The four patterns at a glance

Pattern	What it catches	When to use
`Optional[X]` correctly	Unguarded None access	Any function that may return None
`TypeGuard[X]`	Narrowing failures in helper functions	Validator and predicate functions
`Protocol`	Interface mismatches across component boundaries	Anything using duck typing across modules
`@overload`	Return type ambiguity in multi-dispatch	Functions that return different types for different input types

Python Developer AI Toolkit, Part 2: Five CLI scripts that automate the prompts

Peyton Green — Wed, 22 Apr 2026 17:00:05 +0000

Part 1 of this series covered the prompt library — 272 prompts for Python and backend development organized by task type. This is Part 2: the five CLI scripts that turn those prompts into automation tools you can run from the command line or wire into your development workflow.

The goal isn't to replace your IDE or your existing toolchain. It's to make AI-assisted code review, test generation, documentation, commit messages, and multi-step analysis available as shell commands — so the prompts run when you need them, not when you remember to open a chat window.

All five scripts use the same dependencies (just requests) and support both Claude and GPT via environment variable. They're in the scripts/ directory of the AI Dev Toolkit.

Script 1: ai_code_reviewer.py

What it does: Sends a source file to AI and returns a structured code review.

python ai_code_reviewer.py src/app.py
python ai_code_reviewer.py src/app.py --format markdown > review.md
python ai_code_reviewer.py src/app.py --format json | jq '.issues[]'

The review covers four categories: security issues, performance concerns, style and readability, and actionable suggestions. The structured output is the key part — "review this code" produces whatever the model feels like. A structured prompt with defined categories produces consistent output you can act on.

Three output formats:

text (default): readable console output
markdown: Markdown-formatted output suitable for saving as a file or pasting into a PR comment
json: machine-parseable output for scripting

# Review every changed file before committing
git diff --name-only HEAD | grep '\.py$' | xargs -I {} python ai_code_reviewer.py {}

# Generate markdown reviews for all files in a PR
git diff origin/main --name-only | grep '\.py$' | while read f; do
  python ai_code_reviewer.py "$f" --format markdown > "reviews/$(basename $f .py)-review.md"
done

When I actually use this: Before any PR that touches business logic. Running it on the file before writing the commit message catches issues I'd otherwise miss — especially the edge cases that don't show up in the happy-path test.

Script 2: test_generator.py

What it does: Analyzes a Python source file and generates a pytest test suite for it.

python test_generator.py src/utils.py
python test_generator.py src/utils.py --output tests/test_utils.py
python test_generator.py src/api/routes.py --output tests/test_routes.py --model gpt

The output is runnable pytest code. The generator analyzes function signatures, type hints, and docstrings to infer what to test — including edge cases and error conditions that aren't always obvious when you're writing the function.

# Generate and immediately run the tests
python test_generator.py src/utils.py --output tests/test_utils.py && python -m pytest tests/test_utils.py -v

The generated tests aren't always perfect out of the box. Complex dependency injection, database interactions, and external API calls need adjustments. But for utility functions, data transformations, and validation logic, the first pass is often 80% of what you'd write manually — in about 15 seconds instead of 20 minutes.

The workflow that's changed how I write code: Run test_generator.py on a new module immediately after writing it. The generated tests are a rapid sanity check on whether the function contract matches what I intended. If the generated tests don't make sense for how I expect the function to be used, the function is probably poorly named or badly documented.

Script 3: doc_generator.py

What it does: Generates structured Markdown documentation from Python source files or entire directories.

# Document a single file
python doc_generator.py src/utils.py

# Document an entire package
python doc_generator.py src/ --output docs/

# Generate and redirect to a file
python doc_generator.py src/api/routes.py > docs/routes.md

The output is structured Markdown: module overview, function/class signatures, parameter descriptions, return types, and usage examples. It reads docstrings if they exist and supplements them — or generates from scratch if there are no docstrings.

For packages with clear type hints and function names, the output requires minimal editing. For older codebases with minimal documentation, it produces a first pass that's faster to edit than to write from scratch.

# Document all Python files in a project and write to docs/
find src/ -name "*.py" | while read f; do
  python doc_generator.py "$f" --output "docs/$(dirname ${f#src/})/$(basename $f .py).md"
done

Script 4: commit_message_ai.py

What it does: Generates a conventional commit message from your staged git diff.

# See the suggested message
git add -p
python commit_message_ai.py

# Copy to clipboard
python commit_message_ai.py --copy

# Commit directly
python commit_message_ai.py --apply

# Install as a git hook (runs automatically on every `git commit`)
python commit_message_ai.py --install

The generated messages follow the Conventional Commits format: type(scope): description. The model reads the actual diff — not just the filenames — and produces a message that accurately describes what changed and why.

The --install flag is the useful one. It installs the script as a prepare-commit-msg git hook. After that, every time you run git commit, your editor opens with an AI-generated message pre-filled. Keep it, edit it, or replace it — but you're starting from something specific rather than a blank cursor.

# Install and test
python commit_message_ai.py --install
git add .
git commit  # Your editor opens with: "feat(auth): add JWT refresh token rotation"

Caveat: The hook reads staged changes from git diff --cached. If you stage everything with git add ., the message covers the full diff. If you stage selectively with git add -p, the message covers only what you've staged. Either works — the message reflects what's actually in the commit.

Script 5: prompt_chain.py

What it does: Runs multi-step AI workflows defined in YAML configuration files.

python prompt_chain.py chains/code_review.yaml --input src/app.py
python prompt_chain.py chains/bug_analysis.yaml --input "Login fails on Safari with SSO enabled"
python prompt_chain.py chains/refactor_plan.yaml --input src/legacy_module.py

Single-prompt interactions work well for isolated tasks. Multi-step analysis — where the output of one prompt feeds into the next — is more powerful for complex problems. prompt_chain.py lets you define these pipelines in YAML without writing Python to wire them together.

A simple chain (included in the toolkit at chains/code_review.yaml):

name: "Staged Code Review"
description: "Deep code review: issues first, then fix suggestions, then a summary"
steps:
  - name: "identify_issues"
    prompt: |
      Analyze this code and list every issue you find — bugs, security concerns,
      performance problems, style violations. Be specific and exhaustive.
      Code:
      {{input}}
    temperature: 0.2

  - name: "generate_fixes"
    prompt: |
      For each issue identified in the previous analysis, provide a concrete fix.
      Show the specific change needed, not general advice.
      Issues identified:
      {{identify_issues}}
    temperature: 0.3

  - name: "write_summary"
    prompt: |
      Write a two-paragraph summary of the code review: what the code does well,
      and what needs attention before merging. Be direct.
      Full analysis:
      {{generate_fixes}}
    temperature: 0.4
    max_tokens: 512

Run it:

python prompt_chain.py chains/code_review.yaml --input src/payment_processor.py

The chains directory in the toolkit includes five pre-built chains:

code_review.yaml — the staged review above
bug_analysis.yaml — issue description → root cause hypotheses → debugging steps
architecture_review.yaml — design analysis → trade-offs → recommendations
test_strategy.yaml — module analysis → test cases → mock requirements
refactor_plan.yaml — code smells → refactor candidates → prioritized action plan

You can modify these or build your own. The YAML format supports variable interpolation, step-level temperature and token limits, and referencing any previous step's output in a later prompt via {{step_name}}.

Putting them together

A realistic workflow for a backend Python module:

# After writing a new module
python test_generator.py src/new_feature.py --output tests/test_new_feature.py
python -m pytest tests/test_new_feature.py -v  # check what it caught

# Before committing
python ai_code_reviewer.py src/new_feature.py --format markdown > .review.md
cat .review.md  # address anything critical

# When committing
git add src/new_feature.py tests/test_new_feature.py
python commit_message_ai.py --apply  # generates and commits with AI message

# If the module is going to be maintained by others
python doc_generator.py src/new_feature.py --output docs/new_feature.md

None of these steps are required. Each one is independently useful. The combination is the point — it removes the friction from the parts of Python development that aren't writing the actual logic.

Setup

All five scripts require only requests:

pip install requests

For Claude (default):

export ANTHROPIC_API_KEY=your_key_here

For GPT:

export OPENAI_API_KEY=your_key_here

Switch models at the command line with --model gpt or --model claude.

The scripts are in scripts/ inside the AI Dev Toolkit. The prompt chains are in scripts/chains/. The 272 prompt library from Part 1 is in prompts/, organized by category.

What's in the full toolkit

The AI Dev Toolkit includes:

272 prompts organized across 10 categories (code generation, debugging, architecture, testing, code review, DevOps, documentation, data/SQL, frontend, AI integration)
5 CLI scripts covered in this article
5 pre-built prompt chains for multi-step workflows
Works with Claude and GPT — bring your own API key

Available at kazdispatch.gumroad.com for $29. Less than an hour of a developer consultant's time.

If you found Part 1 useful, Part 2 is the operational half. The prompts give you the right questions; the scripts give you the automation to run them without friction.

Part 1 of this series: Python Developer AI Toolkit, Part 1: How I stopped rewriting the same prompts and packaged 272 that actually work — update URL after Part 1 publishes

Next in the series: pytest fixtures that actually scale — patterns from 2 years of Python CI pipelines

150+ regex patterns for Python developers: stop rebuilding the same wheel

Peyton Green — Mon, 20 Apr 2026 10:00:05 +0000

Every Python developer I know has a folder somewhere called "regex_snippets" or "useful_patterns" or "patterns_to_remember.txt". Mine grew to 47 files over six years. Half of them were wrong. Three of them were the same URL pattern, each slightly different.

The problem with regex isn't that it's hard to write. It's that it's hard to remember. The syntax is compact enough that you forget it between uses. You Google the same log parser every two months. You copy the same email validator from the same Stack Overflow answer you copied from last year.

I finally sat down and went back through two years of production code and pulled out every regex pattern I'd written more than twice. Added the patterns I keep Googling. Got 150+. Organized them by category, added plain-English explanations, documented the edge cases, and packaged it as something I can actually keep open.

That's the Regex Master Pack.

What's in it

8 cheatsheets covering:

Core syntax (characters, quantifiers, anchors, groups, flags — the complete reference on one page)
Lookaheads and lookbehinds — the stuff that trips everyone up
Common gotchas (greedy vs lazy, backtracking, catastrophic patterns, Unicode pitfalls)
Language differences (Python, JavaScript, Go, Java, Rust, PCRE)
Performance guide (writing fast regex, when NOT to use regex)
Regex in CLI tools (grep, sed, awk, ripgrep, perl one-liners)
Regex in editors (VS Code, Vim, JetBrains, Sublime find-and-replace)
Testing and debugging (how to test regex, debug backtracking, workflows)

150+ patterns in 10 categories:

Category	Count
Email, URL, web	18
Dates and times	16
Numbers and currency	15
Phone and address	14
Passwords and auth	10
Code parsing	20
Log parsing	18
Data extraction	15
Text processing	14
DevOps and infra	15

Every pattern includes the regex, a plain-English explanation of each token, example test strings (matches and non-matches), Python usage, and documented edge cases.

5 interactive Python scripts (stdlib only, no pip installs):

regex_tester.py — interactive REPL, paste a pattern and test strings, see matches highlighted with group details
regex_explainer.py — feed in any regex, get a plain-English breakdown of every token
log_parser_generator.py — describe your log format, get a working regex + Python parser
pattern_search.py — search the entire pattern library by keyword or category
regex_quiz.py — 50 progressive challenges to build muscle memory

A few examples

Log parsing: Apache/Nginx combined format

This is the one I rebuild from memory every time I need it. Now I don't:

import re

APACHE_COMBINED = r'^(\S+)\s+\S+\s+(\S+)\s+\[([^\]]+)\]\s+"(\S+)\s+(\S+)\s+(\S+)"\s+(\d{3})\s+(\d+|-)\s+"([^"]*)"\s+"([^"]*)"'

log_line = '192.168.1.1 - frank [10/Oct/2023:13:55:36 -0700] "GET /api/users HTTP/1.1" 200 1234 "https://example.com" "Mozilla/5.0"'

m = re.match(APACHE_COMBINED, log_line)
if m:
    ip, user, timestamp, method, path, proto, status, bytes_, referer, ua = m.groups()
    print(f"{method} {path} → {status}")
    # GET /api/users → 200

The pattern captures: IP, authenticated user, timestamp, HTTP method, path, protocol, status code, bytes transferred, referer, user agent — named groups version also in the pack.

Code parsing: Python function signatures

Useful for static analysis, documentation generators, or any tooling that needs to understand Python code structure:

PYTHON_FUNC = r'^\s*(?:async\s+)?def\s+(\w+)\s*\(([^)]*)\)\s*(?:->\s*([^:]+))?\s*:'

examples = [
    'def simple(x, y):',
    'async def fetch_data(url: str, timeout: int = 30) -> dict:',
    '    def nested(self) -> None:',
]

for line in examples:
    m = re.search(PYTHON_FUNC, line, re.MULTILINE)
    if m:
        name, params, return_type = m.group(1), m.group(2), m.group(3)
        print(f"fn={name}, params={params!r}, returns={return_type!r}")

Handles sync and async, optional return type annotation, indented (nested/method) definitions.

Data extraction: YAML frontmatter

Every static site generator, Jekyll template, and Hugo theme uses this. Here's a pattern that actually handles the edge cases:

YAML_FRONTMATTER = r'^---\s*\n(.*?)\n---\s*\n'

content = """---
title: My Article
tags: python, regex
published: true
---

Article body starts here.
"""

m = re.search(YAML_FRONTMATTER, content, re.DOTALL)
if m:
    frontmatter = m.group(1)
    # Parse the captured YAML string with yaml.safe_load()

The re.DOTALL flag is required — without it, . won't match newlines and the pattern fails on multi-line frontmatter.

DevOps: semantic versioning

SEMVER = r'^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)(?:-(?P<prerelease>[a-zA-Z0-9.-]+))?(?:\+(?P<buildmeta>[a-zA-Z0-9.-]+))?$'

versions = ['1.2.3', '2.0.0-alpha.1', '1.0.0+build.42', '1.0.0-beta+exp.sha.5114f85']

for v in versions:
    m = re.match(SEMVER, v)
    if m:
        print(f"{v} → major={m.group('major')}, pre={m.group('prerelease')}")

Named groups make the captures readable. The pack includes variants for loose semver (accepts v1.2.3 prefix) and range matching (for tools like npm/pip that use >=, ~=, ^).

The part that actually saves time: `regex_explainer.py`

The interactive tools are the piece I use most. regex_explainer.py takes any regex and breaks it down:

$ python scripts/regex_explainer.py '(?:https?://)?(?:www\.)?([^/\s]+)'

Pattern: (?:https?://)?(?:www\.)?([^/\s]+)

  (?:         Non-capturing group start
    https?    Literal 'http', then 's' is optional (? = zero or one)
    ://       Literal '://'
  )?          End non-capturing group, entire group is optional
  (?:         Non-capturing group start
    www\.     Literal 'www' + escaped dot (. in regex matches anything; \. matches only a literal dot)
  )?          End non-capturing group, optional
  (           Capturing group 1 start
    [^/\s]+   One or more characters that are NOT '/' and NOT whitespace
  )           Capturing group 1 end

This is the tool I wish I had when I was learning regex. It's also useful for auditing patterns someone else wrote.

`pattern_search.py` — the reference you'll actually use

$ python scripts/pattern_search.py "log"

Found 18 results in category 'log-parsing':
  [01] Apache/Nginx Combined Log
  [02] Apache/Nginx Common Log
  [03] Syslog RFC 3164
  [04] Syslog RFC 5424
  ...

$ python scripts/pattern_search.py --category "code"

Found 20 results in category 'code-parsing':
  [01] Python Function Definition
  [02] JavaScript/TypeScript Function
  [03] Python Class Definition
  ...

Who this is for

If you write Python and regularly need to parse logs, validate input, extract structured data from text, or build any tooling that touches text at all — this is the reference that replaces the pile of Stack Overflow bookmarks.

Backend developers parsing logs and validating API inputs. DevOps engineers writing grep/sed pipelines. Data engineers cleaning text data. Anyone who uses regex weekly but reaches for Google every time.

The pack

Regex Master Pack on Gumroad → — $19, one-time purchase.

150+ patterns, 8 cheatsheets, 5 Python scripts. Markdown format (works in any editor, terminal, or note-taking app). Patterns also in JSON for programmatic access. Python 3.8+, stdlib only.

30-day refund if it's not useful.

See also: Python Automation Cookbook — 25 production-ready Python scripts for the automation tasks you keep rebuilding.

openai-agents 0.13.x Silently Dropped openai v1 Support — Here's What Breaks

Peyton Green — Fri, 17 Apr 2026 21:00:05 +0000

openai-agents 0.13.2 Silently Dropped openai v1 Support — Here's What Breaks

Status: READY TO STAGE
Written: 2026-03-27 (W100)
Based on: openai-agents 0.13.2 (released 2026-03-26T23:57Z), ADAM W123 landscape scan
Target slot: April 21-23, 2026 (or sooner if v1→v2 friction shows up in community)
Product: AI Dev Toolkit ($29) — https://kazdispatch.gumroad.com/l/zqeopc
Pre-publish checklist:

[x] Replace [POLAR_OR_GUMROAD_LINK] with live product URL (filled 2026-03-27 W126: kazdispatch.gumroad.com/l/zqeopc)
[x] Verify openai-agents still on 0.13.x at time of publish — confirmed 0.13.5 latest as of 2026-04-06 (patch release, no breaking changes; 0.14.0 still pending — nest_handoff_history rename ships there)
[ ] Check if a2a-sdk 1.0.0 has shipped by publish date — if so, add a note to the A2A section
[ ] Check if nest_handoff_history rename has shipped in 0.14.0 — if so, add a second section for that breaking change
[ ] Verify no duplicate coverage from other recent articles
[ ] Stage to Dev.to draft, dispatch kaz to schedule

Article

openai-agents 0.13.2 shipped on March 26th. If you're running openai v1.x in the same environment, your agents are now broken.

No deprecation warning. No migration guide. Just a new dependency requirement in PyPI metadata that says openai<3,>=2.26.0 — and your pip install openai-agents either fails with a conflict error or silently upgrades openai to 2.x and breaks your existing openai client code.

Here's exactly what changed and how to fix it in about 10 minutes.

What Actually Changed

Before 0.13.2, openai-agents was compatible with both openai v1.x and v2.x. Starting with 0.13.2, it requires openai>=2.26.0 and explicitly drops support for v1.

This is relevant if you:

Have a project that pinned openai==1.x.x or openai<2
Inherited a codebase that hasn't been updated since late 2024
Are running openai-agents alongside other packages that still depend on openai v1

The practical effect: running pip install openai-agents --upgrade in an existing environment will now either fail (if pip can't resolve the conflict) or force openai to 2.x — where some of your existing client-side code may stop working.

What openai v2 Actually Changed

The openai v2 library (released November 2024) restructured the client API significantly:

# openai v1 — you might have code that looks like this
import openai

openai.api_key = "sk-..."
response = openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
print(response.choices[0].message.content)

# openai v2+ — the correct pattern
from openai import OpenAI

client = OpenAI(api_key="sk-...")  # explicit client instantiation
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
print(response.choices[0].message.content)

The top-level openai.ChatCompletion.create() call no longer exists in v2. Same with openai.Completion.create(), the old embedding patterns, and the module-level api_key assignment.

If your codebase uses the v1 API directly alongside openai-agents, you have two problems to fix: the dependency conflict and the client code pattern.

The Fast Fix

Step 1: Check where you stand

pip show openai | grep Version
pip show openai-agents | grep Version

If openai is below 2.26.0 and openai-agents is 0.13.2 or newer, you have a conflict.

Step 2: Upgrade openai

pip install "openai>=2.26.0"

Or if you use a requirements file:

# requirements.txt — update this line
openai>=2.26.0

Step 3: Audit your direct openai calls

The most common v1 patterns that break in v2:

# BROKEN in v2
openai.api_key = os.environ["OPENAI_API_KEY"]
response = openai.ChatCompletion.create(...)

# FIXED
from openai import OpenAI
client = OpenAI()  # reads OPENAI_API_KEY from env automatically
response = client.chat.completions.create(...)

# BROKEN in v2
embeddings = openai.Embedding.create(input=texts, model="text-embedding-3-small")
result = embeddings["data"][0]["embedding"]

# FIXED
from openai import OpenAI
client = OpenAI()
response = client.embeddings.create(input=texts, model="text-embedding-3-small")
result = response.data[0].embedding

The pattern is consistent: every top-level openai.SomeThing.create() call becomes client.some_thing.create() on an explicit OpenAI() instance.

openai-agents Code Itself Doesn't Change

The good news: if you're writing openai-agents code (defining agents, tools, handoffs), none of that changes. The openai-agents API is stable across this version bump. The v2 requirement is about the underlying HTTP client, not the agent framework API.

# This openai-agents code works exactly the same before and after 0.13.2
from agents import Agent, Runner

agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant.",
)

result = Runner.run_sync(agent, "What's the weather today?")
print(result.final_output)

The only thing that changed is what version of openai is running underneath.

What's Still Coming in 0.14.0

This isn't the only breaking change on the horizon. The nest_handoff_history parameter is being renamed in an upcoming 0.14.0 release — a separate change that hasn't shipped yet. If you're using that parameter in handoff configurations, watch the changelog.

The Broader Pattern

The openai-agents library moves fast — 0.13.0 through 0.13.2 shipped across four days in late March. If you're depending on it in production, pinning to a specific version and testing upgrades before deploying is worth the five minutes it takes to set up.

A dependency audit tool can flag these conflicts before they hit your CI pipeline. If you're running more than three or four AI libraries in the same environment, you'll hit this pattern regularly.

[The AI Dev Toolkit includes a dependency-audit prompt set that checks for version conflicts across openai, anthropic, and the major agent frameworks — https://kazdispatch.gumroad.com/l/zqeopc]

Summary

openai-agents 0.13.2 (March 26) dropped openai v1 support
Requires openai>=2.26.0 — no deprecation period
Fix: upgrade openai and update any direct v1-style API calls
Your openai-agents agent definitions don't change
nest_handoff_history rename is a separate 0.14.0 story, not yet shipped

DEV Community: Peyton Green

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

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

What Changed in langchain-adk 0.3.14

Building a Multi-Agent Pipeline with langchain-adk

The Framework Portability Test

Observability: Tracing Multi-Agent Systems with LangSmith

When to Use This Pattern

The Bigger Picture

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

What google-adk 2.0 Changes

1. The Workflow Runtime (The Big One)

2. The Event Model Is Completely Different

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

What Breaks in 2.0

Storage Incompatibility (Critical)

Agent API Changes

Session Schema

Python Version Requirement

The Security Note: LiteLLM

Migration Strategy

When to Migrate

What This Means for Your Tooling

Summary

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

The Pattern

What LocalStack Actually Did

What Astral Actually Did (So Far: Nothing)

What This Isn't

What to Actually Do About It

The Python Community Is Actually Good at This

One More Thing

What We Build With

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

The problem with unbounded gather

Bounded concurrency with asyncio.Semaphore

Retry with exponential backoff

Error isolation in batch processing

Progress tracking for long batches

Timeouts and cancellation

Putting it together: a production batch processor

When to use threads instead of async

A checklist for async AI applications

Further reading

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

The problem with freeform JSON parsing

The Pydantic v2 fix

Extracting JSON from model responses

Prompt patterns that produce consistent schema adherence

Handling validation errors gracefully

Nested schemas

Streaming with structured outputs

A complete working pattern

What this gives you that freeform parsing doesn't

Further reading

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

The One Diagram That Settles It

What Each Protocol Handles

A Concrete Example: Both Protocols in One Flow

When to Reach for Each

Migrating from MCP-only to MCP + A2A

The Stack in One View

Authentication in an MCP + A2A Stack

Further Reading

Testing FastAPI Endpoints Without Spinning Up a Server

TestClient: what it actually does

Dependency overrides: the pattern that changes everything

Fixture pattern: TestClient with scoped overrides

Auth: overriding security dependencies

Async endpoints: the one gotcha

What to test vs what to skip

Putting it together: a minimal test setup

Further reading

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

The CLAUDE.md problem

Template 1: Python API / FastAPI

Template 2: Data pipeline

Template 3: LLM agent / AI app

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

Template 5: Multi-service / team

Pattern 1: `Optional[X]` vs `Union[X, None]` — and why you should use neither

Pattern 2: `TypeGuard` for narrowing in conditionals

Pattern 3: `Protocol` for duck typing that mypy understands

Pattern 4: `overload` for functions with multiple valid signatures

The part that actually saves time: `regex_explainer.py`

`pattern_search.py` — the reference you'll actually use