DEV Community: Peyton Green

Stop writing edge case tests. Let Hypothesis find them instead.

Peyton Green — Tue, 14 Apr 2026 17:00:05 +0000

Three years ago I wrote 47 edge case tests for a URL normalizer. Every test followed the same pattern: I thought of a weird input, wrote the test, confirmed the code handled it.

Hypothesis wrote 10,000 edge cases for the same function in 30 seconds and found a bug I'd never have thought to write a test for.

The bug: the normalizer silently returned an empty string when given a URL consisting entirely of whitespace. My 47 tests all used inputs that looked like URLs. Hypothesis found the whitespace case because it's not trying to think of inputs — it's trying to break your function by generating inputs at the boundary of your constraints.

That's the core shift. You stop describing examples and start describing properties.

What property-based testing actually is

A conventional test says: "given this specific input, I expect this specific output."

A property-based test says: "for any input matching this description, this invariant should always hold."

With Hypothesis, you write that invariant, and Hypothesis generates hundreds or thousands of inputs to try to falsify it. If it finds a failing case, it shrinks the input to the minimal example that still fails — so you get a small, debuggable failing case, not just "it broke on a 5,000-character string."

from hypothesis import given, strategies as st

# Conventional test
def test_round_trip_specific():
    assert decode(encode("hello world")) == "hello world"

# Property-based test
@given(st.text())
def test_round_trip_any_string(s):
    assert decode(encode(s)) == s

The second test runs hundreds of times with different inputs. If encode/decode has a bug with unicode characters, empty strings, null bytes, or strings that are exactly 256 characters long, Hypothesis will find it.

Installing Hypothesis (no account required)

pip install hypothesis

That's it. No API key, no signup, no service to configure. Hypothesis is a pure Python library — it generates inputs locally, shrinks failures locally, stores its example database locally. The full power of property-based testing with zero external dependencies.

# If you're using pytest (which you should be):
pip install hypothesis pytest

The @given decorator integrates directly with pytest. Run your test suite the same way you always do — pytest picks up Hypothesis tests automatically.

Writing your first property

The hardest part of property-based testing is the mental shift from "what specific input should I test?" to "what should always be true?"

Three properties that apply to almost every function:

1. Round-trip invariants — encode/decode, serialize/deserialize, compress/decompress

@given(st.text())
def test_json_round_trip(data):
    # Any string that survives JSON serialization should survive a round trip
    import json
    try:
        serialized = json.dumps({"value": data})
        result = json.loads(serialized)
        assert result["value"] == data
    except (ValueError, TypeError):
        pass  # Some strings can't be JSON-serialized; that's expected

2. Idempotency — applying an operation twice produces the same result as applying it once

@given(st.text())
def test_normalize_idempotent(url):
    once = normalize_url(url)
    twice = normalize_url(once)
    assert once == twice

3. Monotonicity — a sort always produces a result no longer than the input, a filter never produces more items than it received

@given(st.lists(st.integers()))
def test_filter_shrinks_list(items):
    result = [x for x in items if x > 0]
    assert len(result) <= len(items)

These are starting points. As you get comfortable, you'll start finding properties specific to your domain — and those domain-specific properties are where property-based testing earns its keep.

Hypothesis strategies: describing your input space

The strategies module (st) is how you describe what kind of inputs Hypothesis should generate. Some useful ones:

from hypothesis import strategies as st

# Basic types
st.integers()                   # any integer
st.integers(min_value=0)        # non-negative integers
st.floats(allow_nan=False)      # floats, excluding NaN
st.text()                       # any unicode text
st.text(alphabet=st.characters(whitelist_categories=('Lu', 'Ll', 'Nd')))  # alphanumeric
st.binary()                     # bytes
st.booleans()

# Collections
st.lists(st.integers())                            # list of integers
st.lists(st.text(), min_size=1, max_size=50)       # bounded list
st.dictionaries(st.text(), st.integers())          # dict with text keys, int values
st.tuples(st.integers(), st.text())                # fixed-structure tuple

# Composing strategies
st.one_of(st.text(), st.none())                    # text or None
st.builds(MyDataClass, name=st.text(), age=st.integers(min_value=0, max_value=150))

The builds strategy is particularly useful — it generates instances of your data classes or Pydantic models by generating each field separately.

from hypothesis import given, strategies as st
from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

@given(st.builds(User, name=st.text(min_size=1), age=st.integers(min_value=0, max_value=150)))
def test_user_serialization_round_trip(user):
    assert User.model_validate_json(user.model_dump_json()) == user

The shrinking superpower

When Hypothesis finds a failing input, it doesn't report the raw generated input. It shrinks it — tries progressively simpler inputs until it finds the minimal case that still triggers the failure.

This is the feature that makes property-based testing practical. Without shrinking, a failure report might be "failed on a 3,000-character string with these properties." With shrinking, it's "failed on the string '\x00'."

Falsifying example: test_normalize_idempotent(
    url='',  # <-- Hypothesis shrunk this to the minimal failing case
)

You don't need to do anything to enable shrinking — it's automatic for all built-in strategies. Custom strategies built with st.composite and standard combinators also shrink automatically.

The example database

Hypothesis stores failing examples in a local .hypothesis/ directory. The next time you run the test suite, Hypothesis tries those failing cases first, before generating new ones. This means:

Once you've found a bug, the regression test for that bug is implicit — Hypothesis will always try the minimal failing input on future runs.
After you fix the bug and the test passes, Hypothesis removes the example from the database.

Commit .hypothesis/ to version control to share failing examples across the team.

# .gitignore — do NOT add .hypothesis to this
# .hypothesis/  <-- leave this line out

Integrating with your existing pytest suite

Hypothesis tests look like pytest tests with an extra decorator. The integration is seamless:

# test_normalizer.py

import pytest
from hypothesis import given, settings, strategies as st
from myapp.normalizer import normalize_url

# Conventional test — still useful for documenting expected behavior
def test_normalize_removes_trailing_slash():
    assert normalize_url("https://example.com/") == "https://example.com"

# Property test — finds the bugs the conventional tests miss
@given(st.text())
def test_normalize_idempotent(url):
    """Normalizing twice should produce the same result as normalizing once."""
    assert normalize_url(normalize_url(url)) == normalize_url(url)

@given(st.text(min_size=1))
def test_normalize_never_empty_on_nonempty_input(url):
    """A non-empty input should never produce an empty normalized URL."""
    result = normalize_url(url)
    # It's okay to return a default URL, but not an empty string
    assert result != ""

Run with pytest as usual. Hypothesis tests are automatically collected and run.

Settings: controlling how hard Hypothesis tries

By default Hypothesis runs each test with a varying number of examples (roughly 100). You can tune this:

from hypothesis import given, settings, strategies as st

@settings(max_examples=1000)   # Try 1,000 examples instead of 100
@given(st.text())
def test_important_property(s):
    ...

@settings(max_examples=50)    # Faster, for properties you're less worried about
@given(st.lists(st.integers()))
def test_basic_property(items):
    ...

In CI, you might want to run with max_examples=500 for important functions and the default elsewhere. The settings can also be configured globally via environment variable or profile.

# conftest.py — set a default for all Hypothesis tests in this suite
from hypothesis import settings
settings.register_profile("ci", max_examples=500)
settings.register_profile("local", max_examples=100)

import os
settings.load_profile(os.getenv("HYPOTHESIS_PROFILE", "local"))

Then in CI: HYPOTHESIS_PROFILE=ci pytest

Where property-based testing earns its keep

Property-based testing is not a replacement for conventional tests. The combination is stronger than either alone:

Use conventional tests for:

Documenting expected behavior with specific examples
Testing known edge cases you've already thought of
Regression tests for specific bugs (the bug had a specific input — document it)

Use Hypothesis for:

Functions that transform or process data (parsers, normalizers, serializers)
Functions with invariants that should hold across all inputs (sort stability, round-trip correctness)
Functions at system boundaries where the input space is large or unpredictable
Finding the bugs you didn't know to look for

The ratio I've landed on: write the conventional tests first to document behavior, then add one or two @given tests per function that exercise an invariant. The Hypothesis tests don't replace the conventional tests — they find the failures the conventional tests can't anticipate.

The three-line addition that catches the most bugs

If you do nothing else with Hypothesis, add this pattern to your parsers and data-processing functions:

@given(st.text())
def test_does_not_crash(s):
    """The function should handle any input without raising an unexpected exception."""
    try:
        result = my_function(s)
        # If it returns normally, the result should be a valid type
        assert isinstance(result, (str, type(None)))
    except ValueError:
        pass  # ValueError is expected for invalid inputs
    except Exception as e:
        # Anything else is a bug
        raise AssertionError(f"Unexpected exception for input {s!r}: {e}") from e

This doesn't assert correctness — it just asserts that the function doesn't blow up with an unhandled exception on arbitrary input. It catches a surprising number of bugs in functions that were "only ever called with valid data" — right up until they weren't.

What you get: the actual coverage difference

My 47-test URL normalizer test suite covered 47 inputs. Running it with Hypothesis at the default 100 examples means roughly 150 inputs total: my 47 plus 100 generated ones.

That's not the point. The 100 generated inputs aren't random — Hypothesis specifically targets:

Empty strings
Strings with only whitespace
Very long strings
Strings with unicode characters outside ASCII
Strings with null bytes
Strings that look like numbers
Strings at exact power-of-two lengths

It targets the boundary conditions that matter for a text-processing function. My 47 tests didn't include any of those, because I was writing example inputs, not adversarial ones.

The whitespace bug I mentioned at the start? Hypothesis found it in the first run. It had been there for eight months.

Getting started (30 minutes)

Pick one function in your codebase that:

Takes a string or collection as input
Returns a transformed version or True/False
Has an invariant you can state in one sentence ("it should always return a non-empty string", "the output should be a subset of the input", "applying it twice should equal applying it once")

Write the Hypothesis test for that invariant. Run it. See what Hypothesis finds.

The learning curve is the mental shift from example thinking to property thinking. Once you've done it once, you start seeing properties everywhere.

pip install hypothesis

No account. No API key. No service to configure. Local, fast, and it catches the bugs you didn't know to look for.

Part 1 of this series: LocalStack Now Requires an Account — Here's How to Test AWS in Python Without One

Part 2 of this series: pytest fixtures that actually scale — coming April 7

The Automation Cookbook ($39) includes a companion section on test automation patterns — pytest fixtures, mocking strategies, and CI pipeline setup. Available on Gumroad.

Claude 3 Haiku Is Being Deprecated April 19. Here's How to Find and Fix Every Reference in Your Python Code.

Peyton Green — Sun, 12 Apr 2026 06:58:01 +0000

If you've been building with the Anthropic API, you have 26 days to audit your Python code for one specific string: "claude-3-haiku".

On April 19, 2026, Anthropic is deprecating the Claude 3 Haiku model family. Any API call that hardcodes claude-3-haiku-20240307 (the full model ID) or uses claude-3-haiku as a prefix will break. This isn't a soft sunset where the old model keeps running in the background — it's a hard deprecation.

Here's how to find every reference and migrate before the deadline.

Step 1: Find every hardcoded model reference in your codebase

The problem with model strings is they end up everywhere: environment variable defaults, test fixtures, config files, utility functions, half-finished scripts. Start with a ripgrep scan:

# Find every claude-3-haiku reference across your entire project
rg "claude-3-haiku" --type py --type json --type yaml --type env -n

If you don't have ripgrep, standard grep works too:

grep -r "claude-3-haiku" . --include="*.py" --include="*.json" --include="*.yaml" --include="*.env" -n

Common places this shows up that get missed:

# Hardcoded in function defaults
def get_client(model="claude-3-haiku-20240307"):  # ← will break
    return anthropic.Anthropic()

# Buried in config dicts
MODELS = {
    "fast": "claude-3-haiku-20240307",    # ← will break
    "smart": "claude-3-opus-20240229",
}

# In test fixtures
@pytest.fixture
def haiku_client():
    return anthropic.Anthropic(), "claude-3-haiku-20240307"  # ← will break

# In .env files (easy to forget)
DEFAULT_MODEL=claude-3-haiku-20240307  # ← will break

Step 2: Choose your migration target

Claude 3 Haiku was the go-to model for high-volume, latency-sensitive tasks — fast responses, low cost, good for classification, routing, and short generations. Your migration options:

Option A: Claude Haiku 4.5 (like-for-like replacement)

# Before
model = "claude-3-haiku-20240307"

# After
model = "claude-haiku-4-5-20251001"

Claude Haiku 4.5 is the direct successor — same speed profile, same cost bracket, better capability. This is the right call if you're optimizing for throughput or cost.

Option B: Claude Sonnet 4.6 (if you've been underserved by Haiku)

# Before
model = "claude-3-haiku-20240307"

# After
model = "claude-sonnet-4-6"

With 1M context and significantly stronger reasoning, Sonnet 4.6 is worth considering if your Haiku usage was doing light reasoning tasks that occasionally felt underpowered. The cost difference may be acceptable if you're running lower volumes.

Rule of thumb: If you chose Haiku for speed/cost, go to Haiku 4.5. If you chose Haiku because Sonnet felt like overkill, re-evaluate — Sonnet 4.6 at 1M context changes what "overkill" means.

Step 3: Centralize the model string (so this never happens again)

The real problem isn't the migration — it's having model strings scattered across 12 files. Fix the pattern while you're in there:

# models.py — single source of truth
class Models:
    # Current — update here and nowhere else
    FAST = "claude-haiku-4-5-20251001"
    STANDARD = "claude-sonnet-4-6"
    POWERFUL = "claude-opus-4-6"

    # Deprecated — kept for reference, remove after migration
    # HAIKU_3 = "claude-3-haiku-20240307"  # deprecated April 19, 2026

Then in your code:

from models import Models

client = anthropic.Anthropic()
response = client.messages.create(
    model=Models.FAST,  # ← change this one constant, affects all callers
    max_tokens=1024,
    messages=[{"role": "user", "content": prompt}]
)

One file, one change, zero scrambling at the next deprecation.

Step 4: Test before April 19

Don't assume a string swap is safe. Run your test suite with the new model string before the deadline — especially if you have:

Token count assumptions: Response lengths may differ between models
Latency-sensitive code: Haiku 4.5 is fast, but measure your actual p95 if you have SLA requirements
Structured output parsing: Newer models are more consistent, but validate your JSON/Pydantic schemas still parse correctly

A quick smoke test for each Haiku usage in your codebase:

import anthropic

def test_migration_smoke():
    """Verify new model ID works before deadline."""
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=50,
        messages=[{"role": "user", "content": "Say 'migration successful' and nothing else."}]
    )
    assert "migration successful" in response.content[0].text.lower()

Quick checklist

[ ] Run rg "claude-3-haiku" across your entire project (Python, JSON, YAML, .env files)
[ ] Choose migration target: Haiku 4.5 (speed/cost) or Sonnet 4.6 (capability)
[ ] Centralize model strings into a single constants file
[ ] Update all references
[ ] Run test suite with new model IDs
[ ] Deploy before April 19

One more thing: find model strings in AI-generated code

If you use LLMs to generate code — including code review, test generation, or scaffolding — there's a good chance some of that generated code hardcoded claude-3-haiku-20240307 based on training data. Worth a dedicated grep pass on any code that was AI-generated or recently added to your project.

The Python AI Dev Toolkit includes a set of code review prompts specifically for auditing AI model references and configuration strings — useful if you're doing this across a large codebase or multiple projects.

April 19 deadline. Audit takes 10 minutes. Don't get caught with a broken prod deploy.

FastAPI + MCP: Adding Real OAuth 2.1 Auth to Your Python MCP Server

Peyton Green — Fri, 10 Apr 2026 10:25:02 +0000

In the nine days after the MCP Dev Summit, NVD recorded 20 new MCP CVEs. Auth validation failures are the dominant pattern.

Two examples: CVE-2025-6514 — command injection in mcp-remote, CVSS 9.6, 500,000 downloads. CVE-2026-32211 — the Azure MCP Server's SSE transport had zero authentication. Attack chain: enumerate tools, call the shell-passthrough tool (azmcp-extension-az), write a script to ~/.bashrc, exfiltrate Entra ID credentials. Full tenant takeover. Microsoft CVSS: 9.1 CRITICAL. Root cause: CWE-306 — Missing Authentication for Critical Function. (NVD scores it 7.5 HIGH, reflecting only the confidentiality vector; Microsoft's score adds the integrity impact of the full tenant compromise.)

Twenty CVEs in nine days. Auth isn't optional hardening for MCP servers.

The summit ran April 2–3. Six sessions dedicated to authentication. Aaron Parecki — OAuth 2.1 spec author, Director of Identity Standards at Okta — headlined one of them: "Evolution, Not Revolution: How MCP Is Reshaping OAuth." The consistent message from that track: standard OAuth 2.1 done right, not a new scheme invented for MCP.

Of 518 audited production MCP servers, 41% have zero authentication. Not weak auth — none at all. Not because it's hard. The Python SDK has shipped full OAuth 2.1 support since v1.21. The spec is stable. The problem is that nobody connected mcp.server.auth to how Python developers actually build APIs.

This article does that.

What MCP Auth Actually Looks Like (The Quick Version)

Before we get into code: MCP uses OAuth 2.1. Not an MCP-specific auth scheme — standard OAuth 2.1 with PKCE. If you've implemented OAuth before, most of this will be familiar.

The Python SDK's mcp.server.auth module handles:

Authorization server metadata (RFC 8414)
Authorization endpoint (redirects for browser-based flows)
Token endpoint (exchange code for access token, refresh tokens)
PKCE (Proof Key for Code Exchange — required in OAuth 2.1)
Dynamic Client Registration (RFC 7591)

What you provide:

The actual user authentication (how you verify a user is who they say they are)
The authorization decision (which clients can access which resources)
Token storage

That boundary — SDK handles protocol, you handle identity — is the key to making this not terrible to implement.

The Setup

pip install "mcp[auth]>=1.27.0" fastapi uvicorn python-jose[cryptography] passlib[bcrypt]

We'll build:

A FastAPI app as the OAuth 2.1 authorization server
An MCP server with authentication required
A simple client that goes through the OAuth flow

The MCP Server with Auth

Start with the MCP server. This part is almost entirely handled by the SDK:

# mcp_server.py
from mcp.server import Server
from mcp.server.auth import OAuthAuthorizationServerProvider, AuthSettings
from mcp.server.fastmcp import FastMCP
from typing import Any

# Define what the auth server needs to know about your MCP server
auth_settings = AuthSettings(
    issuer_url="http://localhost:8000",  # Your FastAPI app's URL
    resource_server_url="http://localhost:8001",  # This MCP server's URL
    client_registration_options=ClientRegistrationOptions(
        enabled=True,
        valid_scopes=["read", "write"],
        default_scopes=["read"],
    ),
)

mcp = FastMCP(
    "My Protected MCP Server",
    auth=auth_settings,
)

@mcp.tool()
async def get_data(query: str) -> str:
    """Retrieve data — requires authentication."""
    return f"Protected data for query: {query}"

@mcp.tool()
async def write_data(content: str) -> str:
    """Write data — requires write scope."""
    return f"Wrote: {content}"

The auth_settings tells the MCP server to require OAuth 2.1 tokens and where to find the authorization server. The SDK enforces this — unauthenticated requests get a 401.

The FastAPI Authorization Server

This is the part nobody has documented. Here's how to wire FastAPI as the OAuth 2.1 authorization server that the MCP SDK expects:

# auth_server.py
from fastapi import FastAPI, Depends, HTTPException, Request, Form
from fastapi.responses import RedirectResponse, JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional
import secrets
import time

app = FastAPI(title="MCP Authorization Server")

# --- Config ---
SECRET_KEY = secrets.token_hex(32)  # In production: load from env, don't rotate
ALGORITHM = "RS256"  # OAuth 2.1 recommends RS256 or ES256
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# In-memory stores — replace with your database
users_db: dict = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("secret"),
        "scopes": ["read", "write"],
    }
}

# Auth code store: code -> {client_id, redirect_uri, scope, code_challenge, user_id, expires}
auth_codes: dict = {}

# Token store: access_token -> {user_id, scopes, expires}
active_tokens: dict = {}


# --- OAuth 2.1 Discovery Endpoints ---

@app.get("/.well-known/oauth-authorization-server")
async def authorization_server_metadata():
    """RFC 8414 — Authorization Server Metadata.

    The MCP SDK fetches this to discover all OAuth endpoints.
    Every field here must be accurate or the SDK will reject the server.
    """
    return {
        "issuer": "http://localhost:8000",
        "authorization_endpoint": "http://localhost:8000/oauth/authorize",
        "token_endpoint": "http://localhost:8000/oauth/token",
        "registration_endpoint": "http://localhost:8000/oauth/register",
        "scopes_supported": ["read", "write"],
        "response_types_supported": ["code"],
        "grant_types_supported": ["authorization_code", "refresh_token"],
        "code_challenge_methods_supported": ["S256"],  # PKCE required in OAuth 2.1
        "token_endpoint_auth_methods_supported": ["none"],  # Public clients (PKCE)
    }


# --- Dynamic Client Registration (RFC 7591) ---

@app.post("/oauth/register")
async def register_client(request: Request):
    """MCP clients register themselves before the first auth flow.

    RFC 7591 Dynamic Client Registration — the SDK calls this automatically.
    """
    body = await request.json()

    client_id = secrets.token_urlsafe(16)
    # Public clients don't have secrets — PKCE handles security

    return {
        "client_id": client_id,
        "client_id_issued_at": int(time.time()),
        "redirect_uris": body.get("redirect_uris", []),
        "grant_types": ["authorization_code"],
        "response_types": ["code"],
        "scope": "read write",
        "token_endpoint_auth_method": "none",
    }


# --- Authorization Endpoint ---

@app.get("/oauth/authorize")
async def authorize(
    response_type: str,
    client_id: str,
    redirect_uri: str,
    scope: str = "read",
    state: Optional[str] = None,
    code_challenge: str = "",
    code_challenge_method: str = "S256",
):
    """The user-facing authorization page.

    In a real app: show a login form, check session cookies, etc.
    Here we return a simple redirect with a login prompt in the URL.
    """
    if response_type != "code":
        raise HTTPException(400, "Only authorization_code flow supported")
    if code_challenge_method != "S256":
        raise HTTPException(400, "Only S256 PKCE supported (required by OAuth 2.1)")

    # Store the pending authorization — we'll complete it after login
    pending_id = secrets.token_urlsafe(16)
    auth_codes[f"pending_{pending_id}"] = {
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "scope": scope,
        "state": state,
        "code_challenge": code_challenge,
        "expires": time.time() + 300,
    }

    # In production: redirect to your actual login UI
    # Here: redirect to a simple login endpoint
    return RedirectResponse(
        f"/oauth/login?pending={pending_id}",
        status_code=302
    )


@app.post("/oauth/login")
async def complete_login(
    pending: str,
    username: str = Form(...),
    password: str = Form(...),
):
    """Complete the auth flow after user logs in."""
    pending_data = auth_codes.get(f"pending_{pending}")
    if not pending_data or time.time() > pending_data["expires"]:
        raise HTTPException(400, "Invalid or expired authorization request")

    # Verify credentials
    user = users_db.get(username)
    if not user or not pwd_context.verify(password, user["hashed_password"]):
        raise HTTPException(401, "Invalid credentials")

    # Issue authorization code
    code = secrets.token_urlsafe(32)
    auth_codes[code] = {
        **pending_data,
        "user_id": username,
        "issued_at": time.time(),
        "expires": time.time() + 60,  # Auth codes expire in 60s
    }
    del auth_codes[f"pending_{pending}"]

    # Redirect back to the MCP client
    redirect = pending_data["redirect_uri"]
    state_param = f"&state={pending_data['state']}" if pending_data.get("state") else ""
    return RedirectResponse(f"{redirect}?code={code}{state_param}", status_code=302)


# --- Token Endpoint ---

@app.post("/oauth/token")
async def token_endpoint(
    grant_type: str = Form(...),
    code: str = Form(None),
    redirect_uri: str = Form(None),
    client_id: str = Form(None),
    code_verifier: str = Form(None),  # PKCE verifier
    refresh_token: str = Form(None),
):
    """Exchange authorization code for access token.

    PKCE verification happens here — this is what makes public clients secure.
    """
    import hashlib, base64

    if grant_type == "authorization_code":
        if not code:
            raise HTTPException(400, "code required")

        code_data = auth_codes.get(code)
        if not code_data or time.time() > code_data["expires"]:
            raise HTTPException(400, "Invalid or expired authorization code")

        # PKCE verification (required in OAuth 2.1)
        if code_data.get("code_challenge") and code_verifier:
            expected = base64.urlsafe_b64encode(
                hashlib.sha256(code_verifier.encode()).digest()
            ).rstrip(b"=").decode()
            if expected != code_data["code_challenge"]:
                raise HTTPException(400, "PKCE verification failed")

        del auth_codes[code]
        user_id = code_data["user_id"]
        scopes = code_data["scope"].split()

    else:
        raise HTTPException(400, f"Unsupported grant_type: {grant_type}")

    # Issue access token
    access_token = _create_access_token(user_id, scopes)

    return {
        "access_token": access_token,
        "token_type": "bearer",
        "expires_in": ACCESS_TOKEN_EXPIRE_MINUTES * 60,
        "scope": " ".join(scopes),
    }


def _create_access_token(user_id: str, scopes: list[str]) -> str:
    payload = {
        "sub": user_id,
        "scopes": scopes,
        "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
        "iat": datetime.utcnow(),
        "iss": "http://localhost:8000",
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")


# --- Token Introspection (the MCP server calls this to validate tokens) ---

@app.post("/oauth/introspect")
async def introspect_token(token: str = Form(...)):
    """RFC 7662 Token Introspection — the MCP server validates tokens here."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return {
            "active": True,
            "sub": payload["sub"],
            "scope": " ".join(payload.get("scopes", [])),
            "exp": payload["exp"],
        }
    except JWTError:
        return {"active": False}

Connecting MCP Server to Auth Server

The MCP server needs to know where to validate tokens. Update mcp_server.py:

# mcp_server.py (complete)
from mcp.server.fastmcp import FastMCP
from mcp.server.auth import AuthSettings, ClientRegistrationOptions, TokenVerifier
import httpx

class IntrospectionTokenVerifier(TokenVerifier):
    """Validates tokens by calling the auth server's introspection endpoint."""

    async def verify_token(self, token: str) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "http://localhost:8000/oauth/introspect",
                data={"token": token},
            )
            data = resp.json()

        if not data.get("active"):
            raise ValueError("Token is not active")

        return {
            "user_id": data["sub"],
            "scopes": data["scope"].split(),
        }

auth_settings = AuthSettings(
    issuer_url="http://localhost:8000",
    resource_server_url="http://localhost:8001",  # This MCP server's own URL
    client_registration_options=ClientRegistrationOptions(
        enabled=True,
        valid_scopes=["read", "write"],
        default_scopes=["read"],
    ),
)

mcp = FastMCP(
    "My Protected MCP Server",
    auth=auth_settings,
    token_verifier=IntrospectionTokenVerifier(),  # Top-level param, not inside AuthSettings
)

@mcp.tool()
async def get_data(query: str) -> str:
    """Retrieve data — requires read scope."""
    return f"Protected data for query: {query}"

Running It

# Terminal 1: Start the auth server
uvicorn auth_server:app --port 8000

# Terminal 2: Start the MCP server
mcp run mcp_server.py --port 8001

Connect with an MCP client (Claude Desktop, or any SDK client):

The client will discover the auth server via /.well-known/oauth-authorization-server
Complete the OAuth flow (authorize → login → token)
All subsequent tool calls will include the Bearer token automatically

Scope Enforcement (The Part Most Guides Skip)

Having auth is one thing. Enforcing scopes per-tool is another. Here's how to check scopes in your tool handlers:

from mcp.server.fastmcp import FastMCP, Context

@mcp.tool()
async def write_data(content: str, ctx: Context) -> str:
    """Write data — requires write scope."""
    # ctx.auth contains the verified token info from TokenVerifier
    user_scopes = ctx.auth.get("scopes", []) if ctx.auth else []

    if "write" not in user_scopes:
        raise PermissionError("write scope required")

    return f"Wrote: {content}"

Scope enforcement at the tool level means different clients can have different permissions — read-only service accounts, admin tools with write access, etc.

What You'd Change for Production

This example is deliberately minimal. In a real deployment:

Use RS256 or ES256 instead of HS256 — asymmetric signing means the MCP server can validate tokens without calling the introspection endpoint (the public key is enough)
Add refresh tokens — access tokens expire; refresh tokens let clients renew without forcing re-login
Replace in-memory stores — use Redis for auth codes/tokens, a proper database for clients and users
Add a real login UI — the /oauth/login endpoint here assumes a form submit; real apps use session cookies, SSO, or IdP delegation (Keycloak, Auth0, etc.)
Configure CORS — MCP clients running in browsers need appropriate CORS headers on all OAuth endpoints
Rate-limit token endpoints — brute-force protection on /oauth/token and /oauth/introspect
Add Resource Indicators (RFC 8707) — The June 2025 MCP spec update requires Resource Indicators to prevent token mis-redemption attacks. A malicious server can't redeem a token scoped to a specific resource against a different resource. Implementation: add "resource_indicators_supported": True to the discovery endpoint, and accept a resource parameter in token requests. For single MCP server deployments (like this example), the attack surface is limited — but if your authorization server issues tokens for multiple resources, this is a required security control.

The OAuth 2.1 protocol layer here is correct and spec-compliant. Item 7 is only required in multi-resource environments — single MCP server deployments can omit it.

Validate Host headers on all MCP endpoints — CVE-2025-66416 (Python MCP SDK, fixed in mcp>=1.23.0) showed that localhost MCP servers without Host header validation are reachable via DNS rebinding from any browser tab. Validate request.headers.get("host") against your expected hostname on every MCP endpoint. This applies to development servers too, not just production. Our pip pin (mcp>=1.27.0) is already above the patched version — but explicit Host header checks are defense-in-depth.
Prefer StreamableHTTP over bare SSE for anything externally accessible — The 2025-2026 MCP CVE cluster includes four vulnerabilities specifically targeting SSE transports, including CVE-2026-32211 (Azure — Microsoft removed the SSE transport entirely in their patch). StreamableHTTP with proper authentication is the hardened path for any MCP server that isn't strictly loopback.

A Note on FastMCP's OAuth Proxy

FastMCP v3.x ships a built-in OAuth proxy. If you're already using FastMCP, you might wonder whether to use that instead of wiring up the auth server manually.

The answer matters because of GHSA-5h2m-4q8j-pqpj — an unresolved security advisory against FastMCP's OAuth proxy as of this writing. The proxy doesn't respect the resource parameter in token requests, and issues tokens scoped to base_url rather than the specific MCP server. In a single-server deployment that's a limited exposure. In a multi-server deployment, it means a token issued for one server can be reused against another — which is precisely the attack that OAuth 2.1's Resource Indicators (RFC 8707) exist to prevent.

The pattern in this article — implementing OAuthAuthorizationServerProvider yourself — doesn't use the FastMCP proxy at all. It issues tokens correctly scoped to the resource server. Use this approach until the FastMCP advisory is resolved.

The 41% Problem

That production audit didn't find 41% zero-auth servers because developers don't care about security. They found it because nobody had written this guide.

The MCP Python SDK has had auth support since November 2025. The spec is stable. The reason 41% of production servers have zero auth is that "add OAuth 2.1 to your FastAPI MCP server" was undocumented, and undocumented things don't get implemented.

Hopefully this helps close that gap.

The patterns above handle auth. The Python MCP Production Server Kit ($79) adds multi-worker deployment (Redis session backend, Kubernetes manifests), a TestClient substitute for pytest — without subprocess spawning — and Prometheus monitoring. Everything you need to run this in production, not just in dev.

The Python Testing Toolkit: 4 Drop-In Files for Production pytest

Peyton Green — Thu, 09 Apr 2026 14:00:06 +0000

Every Python project eventually needs the same test infrastructure. You write conftest.py from scratch, reach for the same Hypothesis patterns, copy async fixture code from a Stack Overflow answer that's three major versions out of date.

I've done this enough times that I started keeping the files. This week I packaged them up.

The Python Testing Toolkit is available now on Gumroad: four production-ready Python files — conftest_production.py, hypothesis_strategies.py, async_test_patterns.py, and parametrize_factories.py. $49, one-time download, MIT license.

Here's exactly what's in each one.

`conftest_production.py` — the conftest you'd write if you had eight hours

Most conftest.py tutorials show you database fixtures or HTTP mocking. Almost none show both in a form that composes cleanly with async tests, FastAPI dependency overrides, and environment variable isolation.

This file handles all of it:

# Transaction rollback per test — no cleanup needed
def test_creates_user(db_session):
    user = User(name="Alice", email="alice@example.com")
    db_session.add(user)
    db_session.flush()

    result = db_session.get(User, user.id)
    assert result.name == "Alice"
    # rolls back automatically — nothing persists between tests

# FastAPI dependency overrides without boilerplate
def test_endpoint_with_mock_service(client_with_override):
    mock_service = MagicMock(return_value={"id": 1, "name": "Alice"})

    with client_with_override({get_user_service: lambda: mock_service}) as c:
        response = c.get("/users/1")
        assert response.status_code == 200

# Environment variables that reset between tests
def test_feature_flag(env_override):
    with env_override(ENABLE_BETA="true", DATABASE_URL="sqlite:///:memory:"):
        result = function_that_reads_env()
        assert result.used_beta_path
    # ENABLE_BETA is unset again here — no state leaks

The database fixture defaults to SQLite with transaction rollback per test. Swap to Postgres by setting a DATABASE_URL environment variable — the fixture handles both without modification. HTTP mocking covers both httpx (via respx) and requests (via responses) because most projects have both.

Drop this into your project root as conftest.py. pytest finds everything automatically.

`hypothesis_strategies.py` — stop using `.text()` for emails

Hypothesis ships with text(), integers(), floats(), dates(). These are correct types for testing algorithms. They're useless for testing application code that expects email addresses, usernames, financial amounts, or URL paths.

When Hypothesis generates "\x00\x7f\ud800" as a test email, your validator fails for the wrong reason. You fix the test to exclude those cases, not the code. The test becomes meaningless.

This file has 30+ strategies for the types that appear in every domain:

from hypothesis import given
from hypothesis_strategies import emails, usernames, monetary_amounts

@given(email=emails(), username=usernames())
def test_user_registration(email, username):
    user = User.create(email=email, username=username)
    assert user.email == email.lower()
    assert len(user.username) >= 3

from decimal import Decimal
from hypothesis_strategies import monetary_amounts, order_data

@given(amount=monetary_amounts(min_value=Decimal("0.01"), max_value=Decimal("10000.00")))
def test_payment_processing(amount):
    result = process_payment(amount)
    assert result.status == "success"
    assert result.charged == amount

# Generate complete valid payloads in one line
@given(registration=user_registration_data())
def test_registration_endpoint(client, registration):
    response = client.post("/register", json=registration)
    # Should be 201 (success) or 422 (validation error), never 500
    assert response.status_code in (201, 422)

The user_registration_data() strategy composes emails(), usernames(), and optionally phone_numbers_e164() into a valid registration dict. order_data() generates line items with Decimal amounts that sum correctly. Each strategy accepts parameters for tightening bounds when you need to test specific edge cases.

The Hypothesis article earlier in this series covers the basics. This file is the production-grade implementation.

`async_test_patterns.py` — the patterns that don't appear in the pytest-asyncio README

Async testing has three problems that show up together:

Your httpx.AsyncClient doesn't know about FastAPI's lifespan (startup/shutdown events)
Background tasks run after the response returns, so your assertions fire before the side effects settle
AsyncMock syntax is verbose and inconsistent across Python versions

# Lifespan-aware async client — startup and shutdown events fire correctly
async def test_create_post(async_http_client):
    response = await async_http_client.post("/posts", json={
        "title": "Hello",
        "body": "World",
    })
    assert response.status_code == 201

# Background tasks: drain before asserting side effects
async def test_welcome_email_sent(async_http_client, drain_background_tasks):
    response = await async_http_client.post("/users", json={"email": "new@example.com"})
    assert response.status_code == 201

    await drain_background_tasks()  # flush FastAPI's background task queue

    assert email_inbox.has_message_for("new@example.com")

# AsyncMock helpers — concise patterns for common cases
from async_test_patterns import async_mock_returning, async_mock_raising

async def test_service_timeout(async_http_client):
    import httpx

    with patch("myapp.external.fetch_data", async_mock_raising(httpx.TimeoutException(""))):
        response = await async_http_client.get("/data")
        assert response.status_code == 503

The file also includes async_db_session (async SQLAlchemy with SAVEPOINT rollback per test), assert_all_resolve() (run a list of coroutines in parallel and assert they all complete within a timeout), and a guide to event loop scope with the actual trade-offs documented.

The most common async test failure — "event loop is closed" — is fixed by adding two lines to pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"

This file includes that config and explains why.

`parametrize_factories.py` — readable parametrize that doesn't collapse under its own weight

@pytest.mark.parametrize with more than four cases becomes a list of tuples that nobody can read six months later. The generated test IDs are test_function[param0-param1-param2], which tells you nothing about which case failed.

from parametrize_factories import Case, cases

@pytest.mark.parametrize("case", cases(
    Case("valid email", value="user@example.com", expected=True),
    Case("no at sign", value="notanemail", expected=False),
    Case("empty string", value="", expected=False),
    Case("missing domain", value="user@", expected=False),
))
def test_email_validation(case):
    assert is_valid_email(case.value) == case.expected

Output:

PASSED test_email_validation[valid email]
PASSED test_email_validation[no at sign]
FAILED test_email_validation[empty string]

# All combinations of dimensions — 9 test cases from 3+3 values
from parametrize_factories import matrix

@pytest.mark.parametrize("combo", matrix(
    method=["GET", "POST", "DELETE"],
    role=["admin", "viewer", "anonymous"],
))
def test_access_control(combo, client):
    response = make_request(method=combo["method"], role=combo["role"])
    assert response.status_code != 500

# Skip cases where environment isn't configured
from parametrize_factories import skip_if_missing
import os

@pytest.mark.parametrize("db_url", [
    skip_if_missing(os.getenv("POSTGRES_URL"), "POSTGRES_URL not set"),
    skip_if_missing(os.getenv("MYSQL_URL"), "MYSQL_URL not set"),
    "sqlite:///:memory:",
])
def test_database_connection(db_url):
    engine = create_engine(db_url)
    ...

No pytest-cases dependency — plain pytest throughout.

What it doesn't include

Not tutorial content. No explanations of what @pytest.fixture does or how scopes work. If you're new to pytest, pytest fixtures that actually scale covers that.

Not Django-specific. The HTTP mocking and parametrize factories work anywhere. The database and async fixtures are FastAPI-oriented. The Hypothesis strategies are framework-independent.

Not a subscription, SaaS integration, or CI dashboard. Four Python files. Download once, use in any project, forever.

Getting it

Python Testing Toolkit on Gumroad →

$49 one-time. MIT license. Includes all four files and the README with drop-in instructions, pyproject.toml configuration, and CI integration notes.

If you've been following the Testing Without the Subscription Tax series, this is the practical companion to the articles. The async testing article drops in September — the async_test_patterns.py file from this toolkit is the implementation behind the patterns in that article.

Questions? Drop them in the comments or reach me at @peytongreen_dev.

Part of the Testing Without the Subscription Tax series.

A2A v1.0.0 Is Live — What Changed and What It Means for Your Python Agents

Peyton Green — Wed, 08 Apr 2026 10:00:00 +0000

The A2A protocol hit v1.0.0 on March 12, 2026. I wrote a quickstart last week — two agents talking to each other locally, under 15 minutes. This is the follow-up: what actually changed in v1.0.0, and what it means for agents going beyond localhost.

The short version: v1.0.0 isn't just a version bump. It landed four things that matter for production deployments: signed agent identity, proper OAuth flows for headless agents, multi-tenancy, and paginated task listing. None of these appeared in v0.x.

The SDK (a2a-sdk on PyPI) is still at v0.3.25 stable. There's a v1.0.0a0 alpha if you want to live on the edge. That gap is actually useful context — it tells you what the spec can do that the SDK hasn't exposed yet, and where to plan ahead.

What v1.0.0 Added

The full changelog is on the a2aproject/A2A GitHub repo. Here's what matters practically.

1. Signed Agent Cards (JWS)

The Agent Card is how one agent discovers another: its name, capabilities, supported input/output types, and endpoint URL. In v0.x, Agent Cards were plain JSON — no authentication of the card itself. Any server could claim to be any agent.

v1.0.0 adds JWS (JSON Web Signature) to Agent Cards. An agent can now cryptographically sign its own card, and a caller can verify the signature before trusting the card.

Why this matters: In a multi-agent system with agents from different teams or vendors, you can't assume every agent card is legitimate. JWS verification gives you a trust root at the identity layer — before any task is delegated.

Implementation sketch (conceptual — SDK alpha required for full support):

import json
from jose import jws, jwk

def sign_agent_card(card: dict, private_key_pem: str) -> str:
    """Sign an Agent Card and return the JWS compact serialization."""
    key = jwk.construct(private_key_pem, algorithm="RS256")
    payload = json.dumps(card).encode()
    return jws.sign(payload, key, algorithm="RS256")

def verify_agent_card(token: str, public_key_pem: str) -> dict:
    """Verify a signed Agent Card and return the card dict."""
    key = jwk.construct(public_key_pem, algorithm="RS256")
    payload = jws.verify(token, key, algorithms=["RS256"])
    return json.loads(payload)

In practice, you'd embed the JWS token in the /.well-known/agent.json response, and clients verify before registering the agent in their registry. The SDK will expose this cleanly once v1.0.0 stable ships — for now, the pyJWT or python-jose approach works against the spec.

2. OAuth 2.0: Device Code Flow + PKCE

v0.x had basic OAuth 2.0 support. v1.0.0 modernized it in two specific ways that matter for agent deployments:

Device Code Flow (urn:ietf:params:oauth:grant-type:device_code): For agents that run headless — no browser, no interactive login, no user present. Instead of redirecting to a login page (which headless agents can't handle), the agent polls a device authorization endpoint while the user approves on a separate device.

import asyncio
import httpx
import time

async def device_code_auth(auth_server: str, client_id: str, scope: str) -> str:
    """Complete the OAuth 2.0 Device Code flow. Returns access token."""
    async with httpx.AsyncClient() as client:
        # Step 1: Request device code
        resp = await client.post(
            f"{auth_server}/device/code",
            data={"client_id": client_id, "scope": scope}
        )
        resp.raise_for_status()
        device_auth = resp.json()

        print(f"Go to: {device_auth['verification_uri']}")
        print(f"Enter code: {device_auth['user_code']}")

        # Step 2: Poll for token
        interval = device_auth.get("interval", 5)
        expires_in = device_auth["expires_in"]
        deadline = time.time() + expires_in

        while time.time() < deadline:
            await asyncio.sleep(interval)
            token_resp = await client.post(
                f"{auth_server}/token",
                data={
                    "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                    "device_code": device_auth["device_code"],
                    "client_id": client_id,
                }
            )
            token_data = token_resp.json()
            if token_data.get("access_token"):
                return token_data["access_token"]
            if token_data.get("error") == "authorization_pending":
                continue
            break

        raise RuntimeError("Device authorization timed out or was denied")

PKCE (Proof Key for Code Exchange): For agents that do use the authorization code flow but can't safely store a client secret — typical in agents deployed as native apps or CLI tools. PKCE replaces the client secret with a one-time verifier/challenge pair generated per-request.

import hashlib
import base64
import secrets

def generate_pkce_pair() -> tuple[str, str]:
    """Return (code_verifier, code_challenge) for PKCE flow."""
    code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode()
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
    return code_verifier, code_challenge

# In authorization request:
verifier, challenge = generate_pkce_pair()
auth_url = (
    f"{auth_server}/authorize"
    f"?response_type=code"
    f"&client_id={client_id}"
    f"&redirect_uri={redirect_uri}"
    f"&code_challenge={challenge}"
    f"&code_challenge_method=S256"
)

# In token exchange:
# Include code_verifier (not the challenge) in the token request

These aren't new OAuth flows — they're RFC 8628 and RFC 7636. What's new is that A2A v1.0.0 specifies them as the expected flows for agent-to-agent authentication, rather than leaving it open.

3. Multi-Tenancy

v0.x A2A assumed one principal per agent endpoint. v1.0.0 adds multi-tenancy: a single A2A server can host multiple isolated tenant contexts, with credentials and tasks namespaced per tenant.

This matters if you're building an A2A agent that multiple organizations or teams use. Without multi-tenancy, you'd run a separate server process per tenant. With v1.0.0, one server handles all of them.

What changes in practice:

# v0.x: one server, one context
from a2a.server.apps import A2AStarletteApplication

server = A2AStarletteApplication(agent=my_agent)
# serve with uvicorn: uvicorn app:server --host 0.0.0.0 --port 8000

# v1.0.0: multi-tenant pattern (SDK support varies — check current docs)
# Tenant context is passed in the task request headers or URL path
# /tenants/{tenant_id}/tasks (path-based) or X-Tenant-ID header

# The agent implementation receives tenant context and scopes its work accordingly
async def handle_task(task: Task, tenant_id: str) -> TaskResult:
    db = get_tenant_db(tenant_id)
    config = get_tenant_config(tenant_id)
    # ... process with tenant-scoped resources

The exact SDK surface for multi-tenancy will stabilize in SDK v1.0.0 stable. For now, if you need it, implement it at the HTTP layer using path prefixes or request headers.

4. `tasks/list` with Filtering and Pagination

v0.x tasks/list returned everything: all tasks for an agent, flat list, no pagination. Fine for local development; unusable at scale.

v1.0.0 adds:

Cursor-based pagination — nextCursor field for stable pagination across pages
Filtering — by status, date range, input type
State filtering — query only submitted, working, completed, or failed tasks

import httpx

async def list_recent_failed_tasks(
    agent_url: str,
    access_token: str,
    cursor: str | None = None
) -> dict:
    """Fetch failed tasks with pagination."""
    params = {
        "status": "failed",
        "pageSize": 50,
    }
    if cursor:
        params["cursor"] = cursor

    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{agent_url}/tasks",
            params=params,
            headers={"Authorization": f"Bearer {access_token}"}
        )
        resp.raise_for_status()
        return resp.json()
        # Returns: {"tasks": [...], "nextCursor": "...", "totalCount": 123}

# Paginate through all failed tasks:
async def get_all_failed_tasks(agent_url: str, token: str) -> list:
    all_tasks = []
    cursor = None
    while True:
        page = await list_recent_failed_tasks(agent_url, token, cursor)
        all_tasks.extend(page["tasks"])
        cursor = page.get("nextCursor")
        if not cursor:
            break
    return all_tasks

5. Error Handling: `google.rpc.Status`

v0.x had minimal standardized error responses. v1.0.0 adopts google.rpc.Status for error payloads, which gives you structured errors with a machine-readable code, a human message, and optional detail objects.

# A v1.0.0 error response looks like:
{
    "error": {
        "code": 9,           # FAILED_PRECONDITION
        "message": "Task input type 'audio/wav' not supported by this agent",
        "details": [
            {
                "@type": "type.googleapis.com/google.rpc.BadRequest",
                "fieldViolations": [
                    {
                        "field": "input.type",
                        "description": "Supported types: text/plain, application/json"
                    }
                ]
            }
        ]
    }
}

# Error handling in your client:
async def delegate_task(agent_url: str, task: dict) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{agent_url}/tasks", json=task)
        if not resp.is_success:
            error = resp.json().get("error", {})
            code = error.get("code")
            message = error.get("message", "Unknown error")
            # google.rpc codes: 0=OK, 3=INVALID_ARGUMENT, 5=NOT_FOUND, 9=FAILED_PRECONDITION, ...
            raise A2AError(code=code, message=message, details=error.get("details", []))
        return resp.json()

Code 9 (FAILED_PRECONDITION) is worth calling out — it's for "task rejected because a precondition wasn't met," which covers the common case of sending the wrong input type to a specialist agent.

The Spec-vs-SDK Gap: What It Actually Means

The A2A protocol spec is at v1.0.0. The a2a-sdk package on PyPI is at v0.3.25 stable, with a v1.0.0a0 alpha.

This isn't a warning sign — it's normal for a protocol-first project. The spec stabilizes first. The SDK catches up. The gap exists because the spec team and the SDK team (even if it's the same people) have different release velocity.

What it means practically:

You can implement v1.0.0 features today by calling the HTTP API directly. The spec is the contract. The SDK is a convenience wrapper. Everything in this article works without the SDK.
The a2a-sdk v0.3.25 still works fine for the quickstart-level patterns — agent discovery, task delegation, message exchange. Those APIs haven't changed.
v1.0.0a0 alpha is available if you want the full SDK surface for signed cards and multi-tenancy. Production caution applies (alpha = API might shift before stable).
SDK v1.0.0 stable will ship — and when it does, the migration from v0.3.25 will be straightforward. The protocol hasn't broken backward compatibility.

Where to Go From Here

The quickstart gives you two agents talking on localhost. This article gives you the production primitives.

The logical next step is putting them together: a signed, OAuth-authenticated, multi-tenant A2A setup that you'd actually deploy. That's more infrastructure than a 15-minute tutorial allows — but the building blocks are here.

If you're at the "I want to actually deploy this" stage, the pieces in order:

Sign your Agent Cards (JWS) so callers can verify identity
Use Device Code flow for headless agents that need to authenticate
Add PKCE to authorization code flows for non-server agents
Implement multi-tenancy at the HTTP layer if you're serving multiple orgs
Use tasks/list pagination for any monitoring or debugging tooling

The protocol is production-ready. The SDK is catching up. That gap is a window.

The AI Dev Toolkit includes prompts for designing agent systems, reviewing A2A implementations, and generating structured task schemas — the kind of thing that's tedious to write from scratch every time you start a new agent project.

pytest fixtures that actually scale: patterns from 2 years of Python CI pipelines

Peyton Green — Tue, 07 Apr 2026 13:30:20 +0000

I spent two years watching the same test suite failures in CI.

Not logic failures — the test logic was fine. Infrastructure failures. The fixture that created a database table but didn't clean it up. The S3 mock that shared state across test files and produced a passing suite locally and a flaky one in CI. The fixture that took 4 seconds to set up because it was creating a full schema, and ran 200 times across the test suite.

The failures had the same root cause every time: fixtures optimized for "it works" rather than for isolation, speed, and reliability under CI's parallel execution model. Once I understood the patterns that caused the problems, I could see them appearing in every codebase I touched.

These are the four patterns I use now. They don't require any new dependencies — just a clear model of how pytest's scope and teardown machinery works.

The scope problem (and why function scope is your default)

pytest fixtures have four scope levels: function, class, module, and session. The default is function. The mistake I see most often is upgrading to module or session scope too aggressively to "speed things up."

The performance gain is real. A session-scoped database connection is faster than recreating it 200 times. The hidden cost is test isolation — the assumption that each test starts with a clean state.

Session-scoped fixtures share state across the entire test run. If test A modifies shared state and test B reads it, you have an ordering dependency. The tests pass when run in isolation and fail when run together — or worse, pass in one order and fail in another. CI parallelism makes this non-deterministic.

The rule I follow:

Expensive, read-only setup → session scope (loading a config file, establishing a connection pool, reading fixture data from disk)
Stateful setup → function scope (anything that creates, modifies, or deletes records)
Explicit reset instead of scope upgrade → always

If you want module or session scope for performance but need clean state per test, yield + teardown is the right tool:

@pytest.fixture(scope="module")
def db_connection():
    conn = create_connection()
    yield conn
    conn.close()  # cleanup after module completes

For stateful resources, function scope + fast setup is better than session scope + flaky isolation:

@pytest.fixture(scope="function")
def clean_table(db_connection):
    """Truncate and repopulate test data for each test."""
    db_connection.execute("TRUNCATE test_records")
    db_connection.execute("INSERT INTO test_records VALUES ...")
    yield db_connection
    # teardown implicit — truncation at start of next test handles it

Layered fixtures: build complexity from simple pieces

The most common fixture anti-pattern I see: a large fixture that sets up everything a test might need, even if the test only needs part of it.

# Anti-pattern: monolithic fixture
@pytest.fixture
def full_environment(aws_credentials, s3_bucket, dynamodb_table, sqs_queue, test_user):
    # 40 lines of setup
    yield FullEnv(s3=s3_bucket, db=dynamodb_table, ...)

The problem: every test that uses full_environment pays the full setup cost, even if it only needs S3. And when the SQS setup fails, it breaks tests that don't use SQS at all.

The alternative: layer fixtures from narrow to broad, and let pytest handle composition:

@pytest.fixture
def aws_credentials(monkeypatch):
    """Minimal AWS credential mocking. Every AWS test needs this."""
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")


@pytest.fixture
@mock_aws
def s3_client(aws_credentials):
    """S3 client. Depends on credentials, nothing else."""
    return boto3.client("s3", region_name="us-east-1")


@pytest.fixture
@mock_aws
def test_bucket(s3_client):
    """Pre-created S3 bucket. Depends on s3_client."""
    s3_client.create_bucket(Bucket="test-bucket")
    return "test-bucket"

Now a test that only needs S3 requests test_bucket. A test that needs DynamoDB requests dynamodb_table. They get exactly what they need, nothing else. pytest resolves the dependency tree automatically.

The benefit: failures are isolated. If SQS setup breaks, only SQS-dependent tests fail. The rest of the suite continues.

The mock boundary problem

With moto (or any mock library), the scope of the mock matters as much as the scope of the fixture.

The @mock_aws decorator intercepts boto3 calls at the function level. If you apply it to the fixture, the mock is only active inside that fixture's setup code:

# Bug: mock_aws only active during fixture setup, not during the test
@pytest.fixture
@mock_aws
def s3_client(aws_credentials):
    client = boto3.client("s3")
    client.create_bucket(Bucket="test-bucket")  # this works
    return client
    # mock_aws context exits here

def test_upload(s3_client):
    s3_client.put_object(...)  # boto3 calls here are NOT mocked — real AWS

The correct pattern is to manage the mock context in the fixture using with or to pass the context manager into the test explicitly:

@pytest.fixture
def s3_mock():
    """Provides an active mock_aws context for the duration of the test."""
    with mock_aws():
        yield


@pytest.fixture
def s3_client(aws_credentials, s3_mock):
    """S3 client within the active mock context."""
    client = boto3.client("s3", region_name="us-east-1")
    client.create_bucket(Bucket="test-bucket")
    return client


def test_upload(s3_client):
    # s3_mock is active for the full duration of this test
    s3_client.put_object(Bucket="test-bucket", Key="key.txt", Body=b"content")
    response = s3_client.get_object(Bucket="test-bucket", Key="key.txt")
    assert response["Body"].read() == b"content"

The s3_mock fixture holds the mock_aws() context open for the entire test. s3_client depends on s3_mock, so pytest ensures the mock is active before setting up the client.

This pattern eliminates the "works locally, fails in CI when test ordering changes" class of failures.

The conftest.py architecture: centralize shareable fixtures

If you have AWS fixtures spread across multiple test files, you're re-creating the same setup code everywhere. Any changes — adding a new region, changing the mock scope, adding cleanup — need to be made in multiple places.

The solution is a conftest.py at the appropriate level. pytest automatically discovers conftest.py files and makes their fixtures available to all tests in the same directory and below.

For a typical project structure:

project/
├── conftest.py          # Project-wide fixtures: aws_credentials, s3_mock, sqs_mock
├── tests/
│   ├── conftest.py      # Test-suite-wide fixtures that need the project root
│   ├── test_ingestion/
│   │   ├── conftest.py  # Module-specific fixtures: pre-populated tables, test data
│   │   └── test_*.py
│   └── test_processing/
│       ├── conftest.py
│       └── test_*.py

The root conftest.py contains the infrastructure fixtures — AWS mocking, database connections, session-scoped expensive setup. The module-level conftest.py files contain domain-specific setup — test data, pre-populated tables, service-specific clients.

This layering means:

Adding a new test file in test_ingestion/ automatically gets the AWS mocks
Changing the mock configuration is a one-file change
Module-specific fixtures don't pollute the namespace of other modules

# conftest.py (project root)
import boto3
import pytest
from moto import mock_aws


@pytest.fixture(scope="function")
def aws_credentials(monkeypatch):
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
    monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")


@pytest.fixture(scope="function")
def aws_mock(aws_credentials):
    """Active mock_aws context for the full test."""
    with mock_aws():
        yield


@pytest.fixture(scope="function")
def s3_client(aws_mock):
    return boto3.client("s3", region_name="us-east-1")


@pytest.fixture(scope="function")
def dynamodb_client(aws_mock):
    return boto3.client("dynamodb", region_name="us-east-1")


@pytest.fixture(scope="function")
def sqs_client(aws_mock):
    return boto3.client("sqs", region_name="us-east-1")

Putting it together: a CI-reliable test

A test that uses all four patterns — function scope, layered fixtures, correct mock boundary, conftest architecture:

# tests/test_ingestion/conftest.py
@pytest.fixture
def ingestion_bucket(s3_client):
    """Pre-created bucket with test prefix structure."""
    bucket_name = "test-ingestion-bucket"
    s3_client.create_bucket(Bucket=bucket_name)
    s3_client.put_object(
        Bucket=bucket_name,
        Key="raw/2026-03-24/event_001.json",
        Body=b'{"event": "test", "timestamp": 1711238400}'
    )
    return bucket_name


# tests/test_ingestion/test_processor.py
def test_processes_event_file(s3_client, ingestion_bucket):
    """Test processes a single event file from S3."""
    # Arrange: file already in bucket via fixture

    # Act
    result = process_s3_event(
        bucket=ingestion_bucket,
        key="raw/2026-03-24/event_001.json",
        s3_client=s3_client
    )

    # Assert
    assert result.status == "processed"
    assert result.event_count == 1

    # Verify side effects in S3
    response = s3_client.list_objects_v2(
        Bucket=ingestion_bucket, Prefix="processed/"
    )
    assert response["KeyCount"] == 1

This test:

Has a clean, isolated S3 environment (function scope)
The mock covers the full test execution (mock boundary correct)
Domain-specific setup is in the module conftest (layered fixtures)
The common AWS mocking is inherited from the root conftest (conftest architecture)

If you run this test 50 times in parallel in CI, it produces the same result every time.

The free conftest.py drop-in

If you're starting from a broken LocalStack setup or rebuilding from scratch, the patterns above are packaged into a drop-in conftest.py covering S3, DynamoDB, SQS, SNS, Lambda, and Secrets Manager. Free download.

Get the free conftest.py → (Kit landing page — coming soon)

The full version with all fixtures, plus 24 more production-ready Python scripts for automation pipelines, is in the Python Automation Cookbook.

Python Automation Cookbook — $39 one-time. 25 production-ready Python scripts.

Tags: #python #testing #pytest #devtools #productivity
Series: Python AWS Testing Without the Subscription Tax — Part 2

pytest 8.x is the current stable release as of early 2026. moto 5.1.22 was released March 8, 2026. All fixture patterns above are tested against both.

MCP Dev Summit 2026: What Actually Changed for Python Developers

Peyton Green — Mon, 06 Apr 2026 14:49:35 +0000

The MCP Dev Summit NYC ran April 2-3. If you're building Python AI tools with MCP, here's the short version of what happened and what it means for your code.

The summit shipped Python SDK v1.27.0 (with OAuth resource validation), established OAuth 2.1 as the auth consensus across the ecosystem, and previewed V2 as in active development — your existing mcp.server.auth code is stable, and the direction is clear.

The longer version is below.

The Python SDK V2 story

The MCP Python SDK had been at v1.26.0 since January 24 — 68 days without a release while the TypeScript SDK shipped multiple updates. The summit was the first public statement of intent for what comes next.

Max Isbey (Anthropic) presented "Path to V2 for MCP SDKs" on April 3. Here's what came out of it:

V2 is in active development on the main branch, with v1.x maintenance on a separate v1.x branch — but no Python V2 release or firm timeline was announced. The most concrete output was v1.27.0 itself (shipped April 2, summit day 1): RFC 8707 resource validation for OAuth clients, an idle timeout for StreamableHTTP sessions, a TasksCallCapability backport from the V2 branch, and conformance test integration from main. The backports signal V2 is far enough along to generate stable features — it's not vapor.

The TypeScript SDK is further along: v2.0.0-alpha.1 and alpha.2 shipped April 1 (the eve of the summit), with Standard Schema support (Zod v4 / Valibot / ArkType compatibility), a refactored TaskManager, deprecated method removal, and the first drop of a new Fastify framework package. Python V2 is clearly still pre-alpha — don't build against a Python V2 release before Q3 at the earliest.

What this means for your code:

No breaking changes to mcp.server.auth were announced. The BearerAuth API (BearerAuthProvider, BearerAuthBackend) is unchanged in v1.27.0 — the relevant PR was not merged, and no V2 auth API changes were publicly announced. The RFC 8707 addition affects the OAuth client side (resource indicator binding), not your server auth code. If you're using mcp.server.auth today, v1.27.0 is a safe update with no rework required.

The practical guidance: if you're writing new MCP server code this week, the current mcp.server.auth patterns are what V2 will normalize — not something to be replaced.

Six auth sessions in two days — what the consensus looks like

Aaron Parecki (OAuth 2.1 spec author, Director of Identity Standards at Okta) headlined the auth track with "Evolution, Not Revolution: How MCP Is Reshaping OAuth." The session title is the key point.

The message from the auth track — consistent across six sessions — was that OAuth 2.1 is the answer for MCP auth, not a new MCP-specific auth scheme. Parecki's pre-summit position has been "don't overthink auth in MCP": use standard OAuth 2.1 with PKCE, implement the existing RFCs correctly, and you're done. Six dedicated sessions devoted to this topic isn't a sign of instability — it's the ecosystem converging on a direction that's been clear in the spec for months. (Specific session content will be confirmed once recordings are indexed; the above reflects the session title framing and the consistent pre-summit position Parecki has held.)

What made this summit unusual: Paul Carleton (Anthropic) ran two sessions — "Cross-App Access for MCP" (XAA/ID-JAG) and a joint session with Twilio on SSO consolidation. Auth was the dominant unresolved production problem in MCP, and Anthropic used the summit to consolidate the community around a direction.

Carleton's cross-app access session covered Cross-App Access (XAA / Identity Assertion Authorization Grant): a pattern that allows AI agents to carry user identity across multiple services without separate per-service logins — SSO for agent workflows. XAA has been in the MCP spec since November 2025 (SEP-990); Okta has a developer playground at xaa.dev, and early production adopters include Automation Anywhere, Boomi, Box, and Glean. For Python developers: XAA is in the spec but not yet in the Python SDK. It's the pattern to plan toward for multi-tenant, multi-service agent deployments — but not a requirement for single-server work today.

The pattern that emerged across all six auth sessions: OAuth 2.1 as the baseline, not the advanced option. If you've been treating OAuth as something to add later, the summit signals that later is now.

OpenAI and Anthropic are building the same thing

The most underreported summit storyline was the cross-ecosystem convergence.

OpenAI's "MCP x MCP" keynote (Nick Cooper, April 3) addressed what's been building quietly: OpenAI Agents SDK v0.13.0 shipped list_resources(), read_resource(), and list_resource_templates() on MCPServer — the MCP Resources API that Anthropic's Python SDK is simultaneously implementing. Two teams from competing platforms presenting the same spec implementation at the same event is as close to an official cross-ecosystem endorsement as you'll see. (Specific session content awaiting public recording; the background SDK state is confirmed from release notes.)

This matters for Python developers because it means the MCP Resources layer is likely to become the stable cross-runtime abstraction. A Python MCP server that exposes resources properly should work with both Claude integrations and OpenAI Agents SDK clients — without modification.

That's a meaningful guarantee, and one of the stronger signals from this summit: the protocol is converging, not fragmenting.

What else was announced

The concrete new technical outputs for Python developers:

Python SDK v1.27.0 (April 2, Day 1): The production-ready 1.x update. Pin to mcp>=1.27.0. Key additions: RFC 8707 OAuth resource validation, StreamableHTTP idle timeout, TasksCallCapability backport.
TypeScript SDK v2.0.0-alpha (April 1): Standard Schema support, Fastify integration, TaskManager refactor. Shows the V2 direction before Python catches up.
MCP Apps (SEP-1865): Interactive UI within AI clients via ui:// URI scheme — official extension. Python SDK support not yet announced; TypeScript and the mcp-ui community package are available now.
XAA at xaa.dev (Okta): Reference implementation for Cross-App Access. Useful for understanding the pattern before the Python SDK implements it natively.
Leitschuh "skeleton key" vulnerability talk: Session confirmed at the summit; no CVE or public writeup at time of writing. Watch this space — disclosure writeups typically follow within 5-7 days of conference presentations.
Enterprise signals: Uber and Duolingo teams presented internal MCP deployments — the technology has crossed from ecosystem to production at scale.

What to do this week

Based on the summit outputs, here's the concrete list for Python developers building with MCP:

If you have existing MCP servers:

Update your pin to mcp>=1.27.0 — no breaking changes, and you get RFC 8707 OAuth resource validation and StreamableHTTP idle timeout.
If you're serving resources (files, database rows, anything with uri addressing), verify your resource implementation against the MCP Resources spec — this is the layer that OpenAI Agents SDK now implements client-side, so compatibility here has real cross-ecosystem value.
Auth is no longer optional for public MCP servers. If you're deploying externally and haven't implemented OAuth 2.1, the summit consensus is clear: add it before V2.

If you're starting new MCP server development:

Build against mcp.server.auth now — no breaking changes were announced and no V2 auth API changes were previewed. The current patterns are what V2 will normalize.
Implement MCP Resources properly from the start — the list_resources()/read_resource() interface is now multi-ecosystem, not Anthropic-specific.
Pin to mcp>=1.27.0 — this is the current stable baseline post-summit.

The one thing that probably doesn't change:

The practical implementation of OAuth 2.1 for MCP servers — PKCE, token introspection, scope enforcement — is stable regardless of SDK version changes. The protocol-level auth spec is what Parecki was presenting, not framework-specific API details. Your OAuth implementation won't break from V2.

The gap that the summit didn't close

The Python SDK V2 date remains unspecified — Q1 2026 was the original target and it passed without a Python alpha. If you're shipping production Python + MCP today, plan for an SDK update in H2 2026, but don't block current work waiting for it.

If you're implementing MCP auth in Python, I have a detailed guide coming April 10: FastAPI as an OAuth 2.1 authorization server for MCP, with PKCE, token introspection, and scope enforcement against a real FastAPI app. The patterns are stable against the summit outputs above — no rework needed. Subscribe below to get it when it drops.

Sign up here — no spam, just Python AI tooling posts.

Build Your First A2A Agent Pair in Python (15 Minutes, No Cloud Required)

Peyton Green — Wed, 01 Apr 2026 10:00:06 +0000

Google's A2A protocol hit v1.0 on March 12, 2026. You've probably seen the announcement — enterprise framing, agent interoperability, Salesforce and SAP and ServiceNow on the logo slide.

What you haven't seen is a tutorial that just works locally, without a Google Cloud account, without paid APIs, without enterprise infrastructure. The official demos are GCP demos. They teach the platform, not the protocol.

This is the other tutorial. You'll build two Python agents that communicate over A2A, run them in two terminal windows, and watch a task travel from dispatcher to executor and back — in under 15 minutes, on localhost.

What A2A Is (30 Seconds)

A2A is a protocol for agent-to-agent task delegation. When your orchestrator agent needs to hand off work to a specialist agent, A2A is how they talk.

The mental model that clicked for me is the HTTP analogy. HTTP didn't create the web — it gave every computer a shared language. A2A does the same thing for agents: any A2A-compliant agent can call any other A2A-compliant agent, regardless of what framework built it or where it's hosted.

It's Apache 2.0 licensed and governed by the Agentic AI Foundation (same neutral body that governs MCP). It is not a Google product. It's an open standard that Google initiated.

A2A and MCP are not competitors. They operate at different layers:

┌─────────────────────────────────────────┐
│            Orchestrator Agent           │
├─────────────────────────────────────────┤
│  A2A: delegates tasks to other agents   │  ← what we're building today
├─────────────────────────────────────────┤
│  MCP: calls tools, reads files, APIs    │  ← what you already have
└─────────────────────────────────────────┘

MCP gives your agent access to external resources. A2A gives your agent access to other agents. Both. Different problems.

What We're Building

Two Python scripts:

stats_agent.py — an A2A-compliant agent server (port 9999) that accepts text and returns word count, average word length, and estimated reading time.
dispatcher.py — a client that discovers the stats agent via its Agent Card and submits a text analysis task.

Total code: ~110 lines. No GCP. No paid APIs. No Docker required.

Install

pip install "a2a-sdk[http-server]" uvicorn httpx

The [http-server] extra pulls in Starlette and sse-starlette for the server side. Requires Python 3.10+.

Versions tested: a2a-sdk==0.3.25, uvicorn==0.42.0, httpx==0.28.1.

Note on SDK versions: pip install a2a-sdk installs 0.3.25, the current stable release. A v1.0.0 alpha exists (1.0.0a0) with breaking changes: proto-based Part types replacing TextPart/FilePart/DataPart, SCREAMING_SNAKE_CASE enums, PascalCase operation names (e.g. SendMessage), gRPC transport added. Stable v1.0 SDK is estimated May–June 2026. This article targets 0.3.25 — if you're following along after stable v1.0 ships, pin to 0.3.25 first, then upgrade with the migration guide. The 0.3.x SDK is not deprecated.

Part 1: The Stats Agent (Server)

Every A2A agent announces itself via an Agent Card — a JSON document served at /.well-known/agent-card.json. This is how other agents discover what your agent can do before they call it.

Create stats_agent.py:

# stats_agent.py
import asyncio
import uvicorn
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, TextPart
from a2a.utils import new_agent_text_message


def analyze_text(text: str) -> str:
    """Returns a plain-text analysis of the given text."""
    words = text.split()
    word_count = len(words)
    if word_count == 0:
        return "Empty input."
    cleaned = [w.strip(".,!?;:\"'") for w in words]
    avg_word_len = sum(len(w) for w in cleaned) / word_count
    reading_time_sec = (word_count / 238) * 60  # avg adult reading speed
    freq: dict[str, int] = {}
    for w in cleaned:
        key = w.lower()
        freq[key] = freq.get(key, 0) + 1
    most_common = max(freq, key=lambda w: freq[w])
    return (
        f"Word count: {word_count}\n"
        f"Average word length: {avg_word_len:.1f} characters\n"
        f"Estimated reading time: {reading_time_sec:.0f} seconds\n"
        f"Most frequent word: '{most_common}'"
    )


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

        user_input = context.get_user_input()
        result = analyze_text(user_input)

        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("cancel not supported")


skill = AgentSkill(
    id="text_stats",
    name="Text Statistics",
    description="Analyzes text and returns word count, average word length, and reading time.",
    tags=["text", "analysis", "nlp"],
    examples=["Analyze this paragraph for me.", "How long is this text?"],
)

agent_card = AgentCard(
    name="Text Stats Agent",
    description="A specialist agent that computes statistics for any block of text.",
    url="http://localhost:9999/",
    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=StatsAgentExecutor(),
        task_store=InMemoryTaskStore(),
    ),
)

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

Three things happening here:

Agent Card declares what this agent can do. Any other A2A agent can fetch this at http://localhost:9999/.well-known/agent-card.json and understand the agent's capabilities before sending a task.
StatsAgentExecutor handles the actual work. execute() is called once per task. TaskUpdater handles the lifecycle: start_work() transitions the task to working, complete() closes it out.
A2AStarletteApplication wires the card and the executor into an HTTP server with all the required A2A endpoints (/, /.well-known/agent-card.json, streaming if enabled).

Part 2: The Dispatcher (Client)

Create dispatcher.py:

# dispatcher.py
import asyncio
import httpx
from a2a.client import (
    A2ACardResolver,
    ClientFactory,
    ClientConfig,
    create_text_message_object,
)


SAMPLE_TEXT = """
The A2A protocol was released as v1.0 on March 12, 2026.
It is an open standard governed by the Agentic AI Foundation,
the same body that governs MCP. A2A handles agent-to-agent
communication; MCP handles agent-to-tool communication.
They are complementary protocols, not competitors.
"""


async def main():
    base_url = "http://localhost:9999"

    async with httpx.AsyncClient() as http:
        # Step 1: Discover the agent via its Agent Card
        resolver = A2ACardResolver(httpx_client=http, base_url=base_url)
        card = await resolver.get_agent_card()
        print(f"Discovered: {card.name}")
        print(f"Skills: {[s.name for s in card.skills]}\n")

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

        message = create_text_message_object(content=SAMPLE_TEXT.strip())

        # Step 3: Iterate response events (Message or (Task, Update) tuples)
        async for event in client.send_message(message):
            if hasattr(event, "parts"):
                # Direct Message response
                for part in event.parts:
                    if hasattr(part.root, "text"):
                        print("Result from Stats Agent:")
                        print(part.root.text)
            elif isinstance(event, tuple):
                task, _ = event
                # Task completed — look for agent reply in history
                if task.history:
                    for msg in task.history:
                        if msg.role.value == "agent":
                            for part in msg.parts:
                                if hasattr(part.root, "text"):
                                    print("Result from Stats Agent:")
                                    print(part.root.text)


asyncio.run(main())

The dispatcher doesn't know anything about the stats agent implementation. It discovers the agent's capabilities from the Agent Card, submits a task, and reads the response. Swap the base_url to point at any A2A-compliant agent and the same dispatcher code works.

Run It

Terminal 1 — start the stats agent:

python stats_agent.py

You should see uvicorn start on port 9999. Verify the agent card is being served:

curl http://localhost:9999/.well-known/agent-card.json

Terminal 2 — run the dispatcher:

python dispatcher.py

Expected output:

Discovered: Text Stats Agent
Skills: ['Text Statistics']

Result from Stats Agent:
Word count: 42
Average word length: 5.6 characters
Estimated reading time: 11 seconds
Most frequent word: 'the'

That's two agents communicating over A2A. The dispatcher found the stats agent by resolving its Agent Card, submitted a task using the standard A2A task schema, and the stats agent processed it through the full task lifecycle (submitted → working → completed).

Part 3: Add Streaming (For Long-Running Tasks)

For tasks that take more than a second or two — a long analysis, a web fetch, an LLM call — streaming lets the dispatcher receive progress updates instead of waiting for a single blocked response.

Change one thing:

In stats_agent.py, enable streaming on the agent card:

capabilities=AgentCapabilities(streaming=True),

The dispatcher code doesn't change. ClientFactory reads the agent card capabilities and automatically uses the streaming transport when the agent advertises it. The same async for event in client.send_message(message) loop handles both streaming and non-streaming responses — streaming delivers incremental (Task, Update) tuples as the task progresses, non-streaming delivers a single Message on completion.

With streaming enabled, your dispatcher gets status updates as the task moves from submitted → working → completed. For a long-running task (an LLM call, a web fetch, a database query), that loop is where you'd display progress to the user or forward partial results downstream.

What You've Built

Two A2A-compliant agents that communicate over an open standard. Nothing here is Google-specific. The stats agent could be replaced by any A2A-compliant agent from any vendor — the dispatcher code doesn't change.

That's the actual value of A2A: your orchestrator can call a Salesforce agent, a self-hosted RAG agent, or a neighbor's Python script — as long as they speak A2A, the handshake is the same.

Connecting to Your Existing Python Scripts

If you have existing Python automation scripts — Boto3 pipelines, data processing tools, Selenium automation — wrapping them as A2A agents takes about 10 additional lines:

class MyExistingScriptExecutor(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()

        params = context.get_user_input()      # parse JSON or plain text
        result = your_existing_function(params) # call your script

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

    async def cancel(self, context, event_queue):
        raise NotImplementedError

Your script is now callable by any A2A-compatible orchestrator — Claude Code, LangGraph, Google ADK, or your own dispatcher.

Going Further: A2A + LangChain (without the boilerplate)

If you're already using LangChain, langchain-adk 0.3.14 (released March 20, 2026) lets you wrap any LangChain agent as an A2A server in one import instead of the full executor/handler boilerplate above:

from langchain_adk.a2a import A2AServer
from langchain_adk.agents.llm_agent import LlmAgent

# Your existing LangChain agent becomes an A2A server
agent = LlmAgent(llm=llm, tools=your_tools)
server = A2AServer(agent=agent, name="My Agent", description="Does stuff")
server.run(port=9999)  # exposes /.well-known/agent-card.json automatically

Your LangChain orchestrator can then call other A2A-compliant agents — whether they're built with Google ADK, Crew.AI, or the raw a2a-sdk you used here — using A2AAgent as a standard LangChain tool.

This is where A2A gets interesting at scale: not "one agent calling one other agent" but an orchestrator that delegates to a dynamic pool of specialists, each discoverable via their Agent Card. langchain-adk uses plain asyncio for orchestration (not LangGraph), so there's no graph state overhead.

A full walkthrough (multi-agent delegation patterns, orchestrator + specialist setup) is in the companion article: LangChain's A2A Integration: Building Multi-Agent Systems in Python Without the Cloud Lock-In.

Python Developer AI Toolkit, Part 1: How I stopped rewriting the same prompts and packaged 272 that actually work

Peyton Green — Tue, 31 Mar 2026 17:00:06 +0000

Every time I started a Python code review or backend debugging session, I'd spend 10-15 minutes writing the prompt. Not the review — the prompt. Then I'd get inconsistent output, tweak the prompt, run it again, get something different.

Same thing with API architecture. Same thing with test coverage gaps. I was spending more time on prompt engineering than on the actual problem.

At some point I started keeping notes. "This phrasing produced good output." "This framing gets consistent code review feedback." Three months in, I had 400+ prompt drafts in a doc.

Then I cleaned them up. Tested each one. Cut the ones that didn't produce reliable output. Organized what was left by task type.

272 prompts survived.

This is the first in a series on AI-augmented Python development workflows. Part 1 covers the prompt library. Part 2 covers the five CLI scripts that run these prompts as automation tools.

What makes a prompt reusable vs. a one-off

The prompts that consistently underperformed had two patterns:

Too vague — "Review this code" gets you whatever the model feels like doing that day
Over-specified for one context — "Review this Python Flask route for SQL injection in a healthcare app" only applies to exactly that situation

The ones that work are specific about task type and output format but generic about context. They force the model into a useful reasoning mode without requiring you to rewrite them from scratch each time.

Example — code review prompt that doesn't work:

"Review my code and tell me what's wrong"

Example — code review prompt that works:

"Review the following code for: (1) logical bugs, (2) edge cases not handled, (3) readability issues, (4) performance concerns. For each issue found, provide: the specific line or section, the problem, and a concrete fix. If no issues exist in a category, say so explicitly."

The second one produces a consistent, structured output you can actually act on. It works whether you're reviewing 10 lines of Python or 200 lines of TypeScript.

Why Python/backend developers specifically

The generic AI prompt market is saturated. Search "ChatGPT prompts for developers" and you get packs of 50 prompts — most of which are AI-generated themselves, never tested against real output. They work tolerably for everything and excellently for nothing.

This toolkit is built around the specific problems that come up repeatedly in Python/backend work: reviewing code for correctness and safety, debugging async behavior, designing REST API contracts, analyzing SQL queries for performance problems, writing pytest fixtures that actually isolate what they're testing. These are not the same problems as "help me write a Python script" — they require different framing, different output structure, and different follow-up patterns.

The difference shows up immediately in output quality. Consider debugging an async issue in Python:

Generic prompt:

"Debug this Python code and tell me what's wrong"

Output tends to be: "I see you're using asyncio. This code might have issues with the event loop. Here are some general async debugging tips..."

Python-specific structured prompt:

"Debug the following async Python code. Identify: (1) any incorrect use of await or missing await calls, (2) potential event loop blocking operations (CPU-bound work, synchronous I/O calls), (3) race conditions in shared state, (4) coroutine lifecycle issues. For each issue found, show the specific line, explain the async behavior causing the problem, and give a corrected version."

Output is: specific line numbers, exact explanation of the event loop blocking, corrected code that preserves the original logic.

The same gap applies to SQL. Generic: "review this query." Specific: "Review the following SQL query for: (1) missing index opportunities given the WHERE and JOIN conditions, (2) N+1 query patterns if this is run in a loop, (3) correctness of the JOIN type against the intended result, (4) any implicit type coercions that could cause performance issues." The structured version consistently surfaces things the generic one misses — particularly implicit type coercions, which are a common source of full table scans that look fast in development and die in production.

Same with pytest. "Write tests for this function" produces tests. "Write pytest fixtures and test cases for this function. Cover: the happy path, boundary conditions on each parameter, invalid input types, any exception paths. Use parametrize where it reduces duplication. Each test name should describe the condition being tested, not the method being called." produces a test file you can actually use.

5 prompts from the toolkit (with actual output comparisons)

1. Debugging — Rubber Duck Prompt

"I have a bug: [describe bug]. Here's the relevant code: [paste code]. Walk me through your reasoning about what's causing this step by step, before suggesting any fixes. List every assumption you're making."

Why it works: forces explicit reasoning before jumping to solutions. The "list every assumption" line catches cases where the model is confidently wrong.

2. Architecture — Tradeoffs Framing

"I'm deciding between [option A] and [option B] for [specific component]. Evaluate each on: (1) scalability at 10x current load, (2) operational complexity, (3) time to implement, (4) reversibility. Don't recommend a choice — give me the tradeoffs. I'll decide."

Why it works: "don't recommend a choice" prevents the model from anchoring on one option. You get a genuine comparison.

3. Documentation — Explanation Layer

"Write documentation for the following function. Include: what it does, what each parameter does and its type, what it returns, edge cases and failure modes, and one usage example. Write for a developer who has never seen this codebase."

Why it works: "developer who has never seen this codebase" forces sufficient context in the output.

4. Test Case Generation

"Generate test cases for the following function. Cover: (1) the happy path, (2) boundary conditions for each parameter, (3) invalid inputs and error handling, (4) any edge cases you can identify from the logic. Write the tests in [framework]. Each test should have a clear name that describes what it's testing."

Why it works: the categories force coverage you'd otherwise miss. "Clear name that describes what it's testing" produces readable tests.

5. Refactoring — Minimum Change Principle

"Refactor the following code to improve readability. Constraints: (1) don't change behavior, (2) don't add features, (3) make the smallest change that meaningfully improves clarity. Explain what you changed and why."

Why it works: without constraints, refactoring prompts produce complete rewrites. The constraint forces targeted improvements.

How to actually integrate these into your workflow

The failure mode with prompt libraries is that they live in a doc you open twice and then forget. You're mid-debugging session, you don't want to context-switch to find the right prompt, so you write a quick one from memory. The quick one is fine. The systematic one would have been better, but you're not going to stop and fetch it.

The five scripts in this toolkit solve that by making the prompts part of your actual command-line workflow. You run them against files the same way you'd run a linter.

Code review on a specific file:

python code_reviewer.py --file auth.py

Outputs a structured markdown review: logical bugs section, edge cases section, readability section, performance section. Each finding includes the line, the problem, and a suggested fix. Takes about 20 seconds. You get the same structured output every time regardless of what's in the file.

Test generation:

python test_generator.py --file utils.py --framework pytest

Generates a test file covering happy path, boundary conditions, invalid inputs, and exception paths. The --framework flag controls whether you get pytest, unittest, or a raw outline. --coverage-level strict adds parametrize decorators and fixtures. The output goes to stdout so you can redirect it or inspect it before writing to a file.

Commit message from diff:

git diff | python commit_message.py

Pipe your diff directly. Outputs a conventional commit message with a subject line under 72 characters and a body summarizing the changes. Useful when you've made 15 small changes and writing the commit message from memory produces something like "fixes and updates."

The pattern across all three: prompts as tools, not reference material. You don't think about which prompt to use — you run the script against the file and it handles the prompt selection internally. The cognitive overhead drops to zero.

The 5 Python scripts in the toolkit

The prompts are paired with 5 automation scripts for common dev workflows:

code_reviewer.py — runs a code review prompt against any file via CLI, outputs structured markdown
test_generator.py — generates test cases for a function with configurable framework and coverage level
doc_generator.py — batch documentation generation for a module or directory
commit_message.py — generates commit messages from git diff output
prompt_chainer.py — chains multiple prompts together for multi-step analysis (e.g., review → suggest refactors → generate tests)

Each script uses argparse, works from the command line, and takes a file path as input. No setup beyond an API key.

What 272 prompts covers

Categories in the toolkit:

Code review (42 prompts) — general review, security review, performance review, readability
Debugging (38 prompts) — error diagnosis, logic tracing, test failure analysis
Architecture (35 prompts) — design decisions, tradeoffs, scalability analysis
Testing (32 prompts) — test case generation, coverage analysis, test naming
Documentation (28 prompts) — function docs, API docs, README generation, inline comments
Refactoring (27 prompts) — targeted cleanup, abstraction identification, naming
Learning (22 prompts) — explain code, explain concept, breakdown pattern
DevOps (20 prompts) — CI/CD configuration, Docker, deployment review
Data (18 prompts) — data pipeline review, schema design, query analysis
API design (10 prompts) — endpoint design, error handling, versioning strategy

Every prompt includes a template with fill-in fields, usage notes on when to use it, and an example of good output.

What I learned about prompting consistency

The biggest surprise from building this library wasn't about instruction precision — it was about context framing.

When I started, I assumed the key to consistent output was precise instructions: specific categories, explicit output format, enumerated requirements. That matters, but there's a second factor that turned out to matter more: telling the model who it's talking to and what they already know.

A prompt that says "explain to a developer who is new to this codebase" reliably produces better output than the same prompt that just says "explain this code." Not marginally better — substantially better. The audience framing changes how the model reasons about what context to include, what to assume, what to define. It's the difference between output you have to edit and output you can paste.

The same effect shows up with role framing. Compare two documentation prompts:

Without role framing:

"Document this function."

Output: a docstring that describes the parameters and return type. Technically correct. Not very useful.

With role framing:

"You are a senior developer documenting this function for a junior developer joining the team next week. Write documentation that explains not just what each parameter does, but why the function works the way it does — what design decision it reflects, and what would break if you changed the interface."

Output: documentation that explains the tradeoffs behind the interface choices, flags the non-obvious behavior, and gives the next developer enough context to modify it safely. Substantially different in usefulness, from a small addition to the framing.

The toolkit applies this consistently — role framing and audience framing in every prompt that benefits from it. It's not visible as a trick; it just shows up in the quality of output.

One-time price, no subscription

272 prompts, each one manually tested against real output before making the cut. Not generated, not scraped — built from three months of actual debugging, review, and documentation sessions.

The toolkit is $29 one-time at https://kazdispatch.gumroad.com/l/zqeopc. Not a monthly subscription — you buy it once and it's yours.

30-day refund if it doesn't deliver value. No questions asked.

What prompts do you keep rewriting? Drop the task type in the comments — debugging async errors, writing test fixtures, reviewing SQL queries — and I'll share the prompt from the toolkit. Best comment gets a free copy.

(Part 2 of this series covers the 5 Python automation scripts that run these prompts as CLI tools — coming next week.)

MCP Dev Summit 2026: What Python Developers Should Actually Pay Attention To

Peyton Green — Sun, 29 Mar 2026 17:00:05 +0000

The first MCP Dev Summit runs April 2-3 in New York, and the sessions that matter most to Python developers aren't the ones getting the headline treatment.

Most of the attention will land on the big-picture stuff: OpenAI presenting alongside Anthropic, enterprise adoption numbers, ecosystem growth metrics. That's all fine. But if you're building Python applications that use MCP servers or clients, the sessions you actually need to care about are the technical ones — specifically the auth architecture sessions and the SDK V2 roadmap.

Here's what to watch for, and why.

The Python SDK has been frozen for 63 days

As of this writing, the MCP Python SDK (mcp on PyPI) has been at v1.26.0 since January 24. The TypeScript SDK has shipped multiple releases in the same period. No one has formally announced a Python v2 roadmap — until now.

Max Isbey (Anthropic) is presenting "Path to V2 for MCP SDKs" at the summit. This is the first public statement of intent for a v2 of the MCP Python SDK. The specific session detail that matters: it may include changes to mcp.server.auth.

If you're using mcp.server.auth in production — or you've read any of the FastAPI + MCP integration guides — you should treat this session as a potential compatibility gate. There's a non-trivial chance that v2 reframes how auth is wired into MCP servers. Not necessarily breaking, but worth verifying against your current implementation.

The practical outcome: after April 3, check the session recording or transcript before pinning your dependencies.

Six auth sessions in two days

This isn't a coincidence. The summit schedule has six dedicated sessions on MCP authentication — which tells you two things:

Auth is the dominant unsolved problem in the MCP ecosystem right now.
The ecosystem is close enough to a solution that Anthropic is ready to present it at a public summit.

Aaron Parecki is attending. If that name sounds familiar, he's the author of the OAuth 2.1 draft specification. His presence at a developer summit about MCP auth means the conversation will be grounded in real spec work, not marketing positioning.

Paul Carleton (Anthropic) is presenting on Cross-App Access (the internally-named XAA/ID-JAG project) — essentially how MCP can enable single sign-on across multiple AI agents and tools. This is the "SSO for agents" problem. If your Python agents need to share identity and authorization context, this session is directly relevant.

The auth story in MCP right now is fragmented: STDIO servers have no auth at all, HTTP servers implement it inconsistently, and most tutorials pretend the problem doesn't exist. The summit looks like the moment Anthropic is trying to close that gap.

OpenAI is in the room

Nick Cooper from OpenAI is presenting a keynote called "MCP x MCP" on April 3. The specific content isn't confirmed yet, but given that OpenAI's agents SDK (openai-agents) has recently added list_resources() and read_resource() for MCP Resources — and the Anthropic Python SDK has a parallel PR pending — there's a reasonable expectation that cross-ecosystem MCP Resource support gets announced or formalized.

Why does this matter for Python developers? Because if MCP Resources land consistently across both the Anthropic and OpenAI SDKs, it opens a new integration pattern: an agent built with anthropic can query context from a server that was designed for openai-agents, and vice versa. The standard is the interop layer.

That's the design intent. Whether the implementation holds up is a different question — and the thing to verify from the session outputs.

What you can do now vs. what to wait on

Do now:

Review your current mcp.server.auth implementation and document the exact version and API surface you're depending on. If you're pinned to a specific version, note why.
If you're using any patterns from the FastAPI + MCP auth guides currently circulating, bookmark them — you'll want to know which ones hold after V2 is specified.
Check your STDIO vs. HTTP split: if you're purely STDIO, the auth sessions are less immediately relevant to you. If you're exposing HTTP MCP endpoints, they're directly relevant.

Wait on:

Don't pin your MCP Python SDK version as a safety measure before the summit. You're currently on a 63-day-old version regardless.
Don't refactor your auth implementation based on summit speculation. Wait for the session outputs.
The list_resources() / read_resource() patterns from the OpenAI agents SDK are pre-summit — don't build production workflows around them until you know what the post-summit spec says.

What comes after

April 4 is when the post-summit analysis starts. A few things I'll be publishing:

A synthesis of what the summit's auth sessions mean for the FastAPI + MCP auth patterns I've been documenting — with specific notes on whether mcp.server.auth changes require any updates to existing implementations. That one publishes April 10.
The Python MCP Server Templates I built for OAuth 2.1, API key, STDIO, SSE streaming, and FastAPI blueprint patterns are already available on Gumroad — those will get a post-summit review pass if the auth spec changes.

In the meantime: if you're building with MCP and want the auth patterns to make sense before the summit, the AI Dev Toolkit has the prompts I use for designing auth flows, writing secure FastAPI middleware, and reasoning through PKCE and token introspection. That part doesn't change regardless of what the summit announces.

If you're planning to watch the summit remotely, the sessions to prioritize are the V2 SDK roadmap and the auth architecture block. Everything else is context. The code impact is in those two.

I'll post a summary of what actually changed on April 4 or 5 once the recordings or transcripts are available.

Claude Code 2.1.85: My Python Project Setup After Six Months of Daily Use

Peyton Green — Sat, 28 Mar 2026 17:00:05 +0000

Claude Code just hit v2.1.85 — it skipped 2.1.82 entirely and pushed three patch versions in roughly 24 hours before promoting 2.1.85 to latest. Whatever was getting fixed, it was getting fixed fast.

I've been using Claude Code as my primary AI coding tool for Python projects for about six months now. The rapid patch cycle is a good moment to write up what I've actually learned — the setup that works, the patterns that save time, and the things that looked useful but weren't.

This isn't a "Claude Code is amazing" post. It's the practical stuff I wish someone had written when I started.

The thing most people skip: CLAUDE.md

Claude Code reads a CLAUDE.md file at the root of your project before every session. Most tutorials mention it briefly and move on. It's actually the highest-leverage configuration you can do.

Without it, Claude Code is making educated guesses about your project. With a good one, it knows exactly how to run your tests, what your linting setup is, which files to avoid, and how you prefer to structure code.

Here's the template I use for Python projects:

# Project: [name]

## Stack
- Python 3.12
- pytest + pytest-cov for testing
- ruff for linting/formatting (replaces black + flake8)
- mypy for type checking (strict mode)
- [any frameworks: FastAPI / SQLAlchemy / etc]

## Commands
- Run tests: `pytest tests/ -x --tb=short`
- Run single test: `pytest tests/test_foo.py::test_bar -v`
- Lint: `ruff check . && ruff format --check .`
- Type check: `mypy src/`
- All checks: `make check`

## Project structure
- `src/[package]/` — main package code
- `tests/` — pytest tests (mirrors src/ structure)
- `scripts/` — one-off scripts, not production code

## Conventions
- Type hints on all public functions
- Docstrings on public classes and non-obvious functions
- Tests use fixtures from conftest.py — check there before creating new ones
- No bare `except:` — always catch specific exceptions

## Don't touch
- `migrations/` — hands off unless explicitly asked
- `.env` files — never read or modify
- `pyproject.toml` dependency versions — propose changes, don't apply

The "Don't touch" section is underrated. Claude Code is eager to help, which means it'll sometimes edit things adjacent to what you asked for. Explicit boundaries save time.

The three commands I use every day

1. claude (plain, in project root)

Opens an interactive session with full project context. This is my primary mode. I describe what I want, it proposes changes, I approve or reject. The approval flow feels slow the first week, then becomes natural — you're reviewing diffs, not babysitting output.

2. claude -p "task description"

One-shot mode. I use this for well-defined tasks where I know exactly what I want:

claude -p "Write a pytest fixture that creates a temp SQLite database for testing, tears it down after each test, and lives in tests/conftest.py"

One-shot is faster than interactive for tasks that are specific enough to describe completely. The key is being precise — vague one-shot prompts produce vague results.

3. claude --continue

Resumes the previous session. More useful than I expected. When I leave something half-done and come back, --continue picks up the thread. It's not perfect context restoration, but it's better than re-explaining everything.

Patterns that actually work

Test-first, always

The single most effective workflow I've found: describe the behavior you want as a test, then let Claude Code implement the function.

Me: Write the test first. I want a function `parse_config(path: str) -> Config`
that reads a YAML file and returns a validated Config dataclass.
The test should cover: valid input, missing required field,
invalid type for a field, and file not found.

Claude: [writes tests/test_config.py]

Me: Now implement parse_config to pass those tests.

This works because tests constrain the solution space. Claude Code is much better at implementing against a spec than guessing what you want.

Ask for the plan before the code

For anything non-trivial, I ask "what's your plan?" before "write the code." This takes an extra exchange but prevents long stretches of code I'd reject immediately.

Me: I need to add rate limiting to our API endpoints.
What's your plan before you write anything?

You'll catch misunderstandings early. "I'd add a Redis-backed rate limiter using..." when you don't have Redis in your stack is much cheaper to correct as text than as 200 lines of code.

Keep sessions focused

Claude Code's context window is large but not infinite, and performance degrades as context fills. I've learned to:

Start a new session for each distinct task
Not drag in unrelated files "just in case"
Use --continue only for the same task, not for adjacent ones

One task per session. It sounds inefficient but it's faster in practice.

The prompt library problem

Claude Code is a tool, not a solution. The quality of what it produces is directly proportional to the quality of what you ask for.

After a few months of using it, I noticed I was writing the same types of prompts repeatedly — for test generation, for refactoring, for API design, for documentation. I started keeping a local file of prompts that had worked well.

That file turned into the AI Developer Toolkit, which I published last week — 272 prompts organized by task type (testing, refactoring, debugging, architecture decisions, code review, etc.). It's what I actually use, not a "1000 ChatGPT prompts" dump.

The relevant point here: the prompts that work best with Claude Code are the ones that give it a clear output format, explicit constraints, and an example or two. Generic prompts get generic code.

What doesn't work

Asking it to "look at the codebase and suggest improvements"

You'll get a list of things that are either obvious, wrong for your context, or technically correct but not worth doing. Claude Code is good at executing specific tasks. It's not good at prioritizing.

Letting it run multiple steps without checkpoints

I have autoAcceptEdits: false in my settings. Some people turn this on for speed. My experience: you end up reviewing three interdependent changes at once instead of one at a time, and when something is wrong it's harder to identify where.

Using it for first drafts of complex architecture

For a simple CRUD endpoint, first draft is fine. For something genuinely complex — a custom event loop, a tricky state machine, distributed coordination — I've had better results sketching the design myself first and using Claude Code to fill in the implementation.

The stable vs latest question

Claude Code ships stable, latest, and next tags via npm. I run latest (currently 2.1.85). stable is still pinned at 2.1.77 — well behind.

For daily use on Python projects, I haven't hit stability issues on latest. If you're running it in CI or as part of automated workflows, I'd be more conservative. But for interactive development sessions, latest is fine.

The 2.1.85 patch cycle

I don't know what's in 2.1.83 through 2.1.85 specifically — the npm metadata doesn't include changelogs. But three patches in 24 hours before promotion suggests something was wrong and then fixed. That pattern is familiar to anyone who's shipped software: sometimes you catch a regression mid-release and hotfix until it's stable.

Whatever it was, the tool feels solid on my current Python projects. I'll update this if anything notable surfaces.

Quick-start CLAUDE.md for a Python package

If you're starting from scratch, here's the minimal version:

# [Package Name]

## Test command
pytest tests/ -x

## Lint command
ruff check . && ruff format --check .

## Python version
3.12

## Don't modify
- pyproject.toml (propose changes only)

Add more as you learn what Claude Code consistently gets wrong for your project. The CLAUDE.md is a living document — I update mine every few weeks.

That's what six months of daily Python use has taught me. The tool is good. The setup matters more than most people realize.

If you're using Claude Code on Python projects and have patterns that work for you, I'd like to hear them in the comments — especially around testing workflows and larger codebases.

Python Automation Cookbook, Part 1: The 25 scripts I reach for every week

Peyton Green — Fri, 27 Mar 2026 10:00:05 +0000

Every project I start has the same 20 minutes of setup.

Write the HTTP client with retry logic. Set up the file watcher. Write the CSV parser. Build the rate limiter. Wire up the scheduler. I know exactly how to do all of it — I've done it fifty times. But I still write it from scratch every time, because I never had a clean canonical version I trusted enough to copy.

This year I stopped doing that. I went back through my last 2 years of projects and pulled out every script that (a) I'd rebuilt more than twice and (b) had survived production without modifications. I got 25 scripts. I cleaned them up, documented them properly, and packaged them.

Here's what's in there and why I made the choices I made.

What "production-ready" means in practice

Tutorial scripts have three failure modes:

No error handling — they work until they don't, with no useful error output
Hardcoded config — you have to edit the file to change any parameter
No CLI — you can only run them from another Python script, not from a shell or scheduler

Every script in the cookbook fixes all three. They use argparse for CLI, handle errors with useful exit codes and messages, and are completely self-contained — no shared state, no internal imports required.

The goal: you should be able to drop any script into a new project, read it in 5 minutes, and run it. Then adapt it if you need to.

The pattern I found across all 25 scripts

When I went back through 2 years of projects to pull the scripts worth keeping, I expected to filter by "does this work." That turned out to be the wrong filter. Plenty of scripts worked. What I actually filtered on was: has this been reused without modification?

The scripts that survived that filter shared four properties. None of them are surprising in isolation. What surprised me was how consistently tutorials and quick solutions violate all four.

1. They fail loudly with useful exit codes and error messages, not silently.

A script that swallows an exception and exits 0 is worse than one that crashes — at least the crash tells you something broke. The production-worthy version catches exceptions at the boundary, logs a message that includes the error type and the value that caused it, and exits with a non-zero code. A cron job that returns exit 1 triggers an alert. One that returns exit 0 silently drops your data with no indication anything went wrong.

2. Every configurable value is a CLI arg with a documented default, never hardcoded.

Hardcoded values look like shortcuts when you're writing the script and look like landmines six months later when someone runs it in a slightly different context. File paths, timeouts, retry counts, output directories — all of these appear as argparse arguments with sensible defaults. If you need to run the same script against a different endpoint or with a longer timeout, you pass a flag; you don't edit the file.

3. They're idempotent — running them twice produces the same result.

This matters most for scripts run by schedulers or CI pipelines, where the same script might run twice due to a retry or a race condition. A script that creates a file should check if it already exists. A script that inserts records should handle duplicates. A script that syncs a directory should be a no-op if the directories are already in sync. Idempotence is not free, but the cost of writing it once is much lower than the cost of debugging a corruption caused by a double-run at 3 AM.

4. They do one thing and expose a clean interface.

The scripts that didn't survive tended to grow. Start as a file downloader, acquire some transformation logic, pick up a notification step, accumulate a few flags that change the behavior in incompatible ways. A script doing four things is four things you can't independently test, reuse, or replace. The ones worth keeping are narrow. The composition of narrow scripts produces complex behavior; the script itself stays simple.

Tutorial code violates all four of these constantly — not because the authors are wrong, but because tutorials optimize for clarity at the moment of reading, not for reliability at the moment of production failure. AI-generated scripts have the same problem: they optimize for plausible-looking output, not for what happens when the rate limiter fires at 2 AM and your CI has no one watching it. Every script in this cookbook was debugged in production before it ended up here. None of them are generated.

4 scripts worth showing in detail

1. HTTP client with retry logic and rate limiting

The problem: most requests implementations handle 200s and 500s. They don't handle rate limits (429), transient failures (503), or connection timeouts gracefully. You find out when something breaks at 2 AM.

What the script does:

Configurable backoff (exponential with jitter, adjustable base/max/multiplier)
Handles 429 with Retry-After header parsing — waits the correct amount, not a fixed sleep
Distinguishes retryable errors (5xx, timeout, connection error) from permanent ones (4xx except 429)
Session reuse for performance
Structured logging of every retry attempt with reason

CLI: python http_client.py --url https://api.example.com/endpoint --retries 5 --timeout 30

# Core retry logic — simplified excerpt
def request_with_retry(url, method="GET", max_retries=3, backoff_base=1.0, **kwargs):
    session = requests.Session()
    last_error = None

    for attempt in range(max_retries + 1):
        try:
            response = session.request(method, url, **kwargs)

            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", backoff_base * (2 ** attempt)))
                logger.warning(f"Rate limited. Waiting {retry_after}s (attempt {attempt+1}/{max_retries})")
                time.sleep(retry_after)
                continue

            if response.status_code >= 500:
                wait = backoff_base * (2 ** attempt) + random.uniform(0, 1)
                logger.warning(f"Server error {response.status_code}. Retrying in {wait:.1f}s")
                time.sleep(wait)
                continue

            response.raise_for_status()
            return response

        except (requests.Timeout, requests.ConnectionError) as e:
            wait = backoff_base * (2 ** attempt) + random.uniform(0, 1)
            logger.warning(f"Connection error: {e}. Retrying in {wait:.1f}s")
            last_error = e
            time.sleep(wait)

    raise last_error or requests.RequestException(f"Max retries exceeded for {url}")

2. File watcher with debouncing

The problem: watchdog is great but every implementation I've seen either (a) fires duplicate events on a single save or (b) doesn't handle the "file replaced" pattern that editors like vim use.

What the script does:

Debounces file events with configurable delay (default 500ms) — one callback per logical change, not per filesystem event
Handles create/modify/delete/move events separately with distinct callbacks
Ignores temp files and swap files by default (configurable patterns)
Recursive or non-recursive watching
Graceful shutdown on SIGINT/SIGTERM

# Debouncing logic — simplified excerpt
class DebouncedHandler(FileSystemEventHandler):
    def __init__(self, callback, debounce_ms=500, ignore_patterns=None):
        self.callback = callback
        self.debounce_delay = debounce_ms / 1000
        self.ignore_patterns = ignore_patterns or [r'.*\.swp$', r'.*\.tmp$', r'~$']
        self._timers = {}

    def _should_ignore(self, path):
        return any(re.match(p, path) for p in self.ignore_patterns)

    def on_modified(self, event):
        if event.is_directory or self._should_ignore(event.src_path):
            return
        self._schedule_callback(event.src_path, event)

    def _schedule_callback(self, path, event):
        if path in self._timers:
            self._timers[path].cancel()
        timer = threading.Timer(self.debounce_delay, self.callback, args=[event])
        self._timers[path] = timer
        timer.start()

3. Paginated API collector

The problem: every API paginates differently. Some use page/per_page, some use offset/limit, some use cursor tokens, some use next_url in the response. Writing a collector for each one is tedious and error-prone.

What the script does:

Supports four pagination strategies: page-based, offset-based, cursor-based, and next-URL-based
Configurable via JSON schema describing the API's pagination parameters
Yields records as they arrive (generator) — works on APIs returning millions of records
Rate limiting between requests
Configurable stop conditions (max records, max pages, or custom predicate)

CLI: python api_collector.py --config api_config.json --output records.jsonl

4. Structured CSV/JSON normalizer

The problem: real-world data is inconsistent. Column names vary across exports. Types need coercion. Nested JSON needs flattening. Nulls are represented as empty strings, "NULL", "N/A", "none", or actual nulls depending on the source.

What the script does:

Configurable field mapping (rename columns via config file)
Type coercion with null handling (configurable null representations)
JSON column flattening with configurable depth limit
Duplicate detection and deduplication
Validation with configurable error handling (skip / raise / log)
Output to CSV, JSONL, or SQLite

CLI: python normalizer.py --input data.csv --config schema.json --output clean.jsonl --on-error log

Full list of 25 scripts

HTTP & APIs (6)

http_client.py — retry logic, rate limiting, session reuse
api_collector.py — paginated API collection (4 pagination strategies)
webhook_sender.py — send webhooks with HMAC signing, retry, and delivery tracking
webhook_receiver.py — receive and validate webhooks (Flask-based, signature verification)
oauth_client.py — OAuth 2.0 client with token refresh
graphql_client.py — GraphQL query client with variable substitution

File Operations (5)

file_watcher.py — file system monitoring with debouncing
batch_processor.py — process files in a directory with parallelism control
file_sync.py — sync two directories (local or S3), with diff-only mode
archive_manager.py — compress/decompress with progress, integrity checking
log_rotator.py — rotate and compress log files with configurable retention

Data (5)

normalizer.py — CSV/JSON normalization with type coercion and mapping
deduplicator.py — deduplicate records by configurable key fields
csv_merger.py — merge multiple CSV files with conflict resolution
json_flattener.py — flatten nested JSON to tabular structure
schema_validator.py — validate records against JSON Schema with detailed error output

Scheduling & Jobs (4)

cron_wrapper.py — wrap any command as a cron job with logging, alerting, and lock files
task_queue.py — simple file-based task queue (no Redis required)
retry_runner.py — run a command with retry logic and configurable backoff
job_scheduler.py — schedule jobs with APScheduler, persistent state

System & Cloud (5)

db_backup.py — backup PostgreSQL/MySQL/SQLite with rotation and S3 upload
s3_sync.py — sync local directory to S3 with change detection
process_monitor.py — monitor a process, restart if dead, alert on anomalies
ssl_checker.py — check SSL certificate expiry across a list of domains
docker_cleanup.py — prune stopped containers, dangling images, unused volumes

Scripts I almost cut (and why they survived)

The hardest part of building the cookbook was the cutting phase. I had more than 40 scripts that fit the "rebuilt twice, survived production" criteria. Getting to 25 meant making the case for each one.

A few were hard to cut because their value only shows up in edge cases — the cases you only hit once, at the worst possible time.

cron_wrapper.py was almost cut because every team already has cron. The counterargument came from a near-miss. A data processing job had been running nightly for three months without issues. One night the job took longer than expected and the next scheduled run started before the first one finished. Both instances wrote to the same output file, the second one truncated it mid-write, and the first one finished writing to a file that was now shorter than it started. The corruption was silent — the file was valid CSV, just missing the last third of the records. We found it four days later when a downstream report came up short. cron_wrapper.py uses a lock file to prevent concurrent execution and alerts on lock contention. The lock behavior isn't useful 99% of the time. It's essential the one time it would have mattered.

process_monitor.py was almost cut because modern infrastructure uses systemd for process management. The counterargument: systemd manages services. process_monitor.py manages scripts that aren't services — a nightly batch job, a file processor that runs on demand, a data sync that needs to run continuously but doesn't warrant a full service definition. For these, you want restart-on-failure and anomaly alerting (CPU spike, memory leak, unusual runtime) without the overhead of writing and maintaining a systemd unit file. The script fills that gap and has been used on three different projects specifically for batch jobs that grew beyond what a bare cron job could handle.

retry_runner.py was almost cut because it seemed redundant with http_client.py. The distinction: http_client.py handles retry logic for HTTP requests specifically. retry_runner.py handles retry logic for arbitrary shell commands — a database migration that occasionally fails on the first attempt due to connection pool exhaustion, a file upload to a flaky external service, a test suite that reliably passes on the second run when given a few seconds after the first. It wraps any command, not just HTTP calls, and that generality turned out to matter.

How to adapt these to your codebase

These are starting points, not drop-in libraries. The distinction matters because "drop-in library" implies a black box — you configure it, call it, and don't need to understand what's inside. That's the wrong model for scripts you're going to run in production on your own infrastructure. You should understand what they do before you trust them.

Every script in the cookbook is under 200 lines, with clear section comments that separate the argument parsing, the core logic, and the error handling. The structure is intentional: it makes the scripts easy to read in one sitting, and easy to find the specific part you want to change.

The most common adaptation pattern: copy the script, read it once to understand the flow, find the 5-10 lines that implement the specific behavior you need, strip everything else, extend from there. You're not supposed to use the full script as-is in every case. You're supposed to extract the part that solves your problem and build on it.

A concrete example: the retry logic in http_client.py — the exponential backoff with jitter, the 429 handling with Retry-After header parsing, the distinction between retryable and permanent errors — has been extracted and adapted for at least three different internal tools this year. One was a Slack API client that needed to handle rate limits differently from the default. One was a webhook sender for an API that returned 503 during deployments and needed more aggressive backoff. One was an internal data pipeline that called a rate-limited external service with custom authentication. The core retry logic was the same in all three; the adaptation was in the error classification and the authentication layer. Starting from the script saved 30-60 minutes of writing and debugging the retry edge cases from scratch each time.

Why one-time price

Most tools that help you automate things are SaaS. Monthly subscription, usage limits, API keys you don't control. For a set of scripts you run on your own infrastructure, that model doesn't make sense.

The cookbook is $39 one-time at https://kazdispatch.gumroad.com/l/hydbq. You get the scripts, no subscription, no usage tracking.

30-day refund if it doesn't deliver value. No questions asked.

What's the automation you keep rewriting? Leave it in the comments — file operations, API calls, scheduling, data pipeline boilerplate — and I'll either point you to the script that covers it or admit it's not in there. If three people ask for the same thing, I'll write it. (#discuss)

(Part 2 covers how to chain these scripts together with prompt_chainer.py — building multi-step automation pipelines that use AI to handle the decisions.)

If you want to know when Part 2 drops — and get the free conftest.py template I use on every Python project — sign up here. No spam, just the occasional Python automation post.

DEV Community: Peyton Green

Stop writing edge case tests. Let Hypothesis find them instead.

What property-based testing actually is

Installing Hypothesis (no account required)

Writing your first property

Hypothesis strategies: describing your input space

The shrinking superpower

The example database

Integrating with your existing pytest suite

Settings: controlling how hard Hypothesis tries

Where property-based testing earns its keep

The three-line addition that catches the most bugs

What you get: the actual coverage difference

Getting started (30 minutes)

Claude 3 Haiku Is Being Deprecated April 19. Here's How to Find and Fix Every Reference in Your Python Code.

Step 1: Find every hardcoded model reference in your codebase

Step 2: Choose your migration target

Step 3: Centralize the model string (so this never happens again)

Step 4: Test before April 19

Quick checklist

One more thing: find model strings in AI-generated code

FastAPI + MCP: Adding Real OAuth 2.1 Auth to Your Python MCP Server

What MCP Auth Actually Looks Like (The Quick Version)

The Setup

The MCP Server with Auth

The FastAPI Authorization Server

Connecting MCP Server to Auth Server

Running It

Scope Enforcement (The Part Most Guides Skip)

What You'd Change for Production

A Note on FastMCP's OAuth Proxy

The 41% Problem

The Python Testing Toolkit: 4 Drop-In Files for Production pytest

conftest_production.py — the conftest you'd write if you had eight hours

hypothesis_strategies.py — stop using .text() for emails

async_test_patterns.py — the patterns that don't appear in the pytest-asyncio README

parametrize_factories.py — readable parametrize that doesn't collapse under its own weight

What it doesn't include

Getting it

A2A v1.0.0 Is Live — What Changed and What It Means for Your Python Agents

What v1.0.0 Added

1. Signed Agent Cards (JWS)

2. OAuth 2.0: Device Code Flow + PKCE

3. Multi-Tenancy

4. tasks/list with Filtering and Pagination

5. Error Handling: google.rpc.Status

The Spec-vs-SDK Gap: What It Actually Means

Where to Go From Here

pytest fixtures that actually scale: patterns from 2 years of Python CI pipelines

The scope problem (and why function scope is your default)

Layered fixtures: build complexity from simple pieces

The mock boundary problem

The conftest.py architecture: centralize shareable fixtures

Putting it together: a CI-reliable test

The free conftest.py drop-in

MCP Dev Summit 2026: What Actually Changed for Python Developers

The Python SDK V2 story

Six auth sessions in two days — what the consensus looks like

OpenAI and Anthropic are building the same thing

What else was announced

What to do this week

The gap that the summit didn't close

Build Your First A2A Agent Pair in Python (15 Minutes, No Cloud Required)

What A2A Is (30 Seconds)

What We're Building

Install

Part 1: The Stats Agent (Server)

Part 2: The Dispatcher (Client)

Run It

Part 3: Add Streaming (For Long-Running Tasks)

What You've Built

Connecting to Your Existing Python Scripts

Going Further: A2A + LangChain (without the boilerplate)

Further Reading

Python Developer AI Toolkit, Part 1: How I stopped rewriting the same prompts and packaged 272 that actually work

What makes a prompt reusable vs. a one-off

Why Python/backend developers specifically

5 prompts from the toolkit (with actual output comparisons)

How to actually integrate these into your workflow

The 5 Python scripts in the toolkit

`conftest_production.py` — the conftest you'd write if you had eight hours

`hypothesis_strategies.py` — stop using `.text()` for emails

`async_test_patterns.py` — the patterns that don't appear in the pytest-asyncio README

`parametrize_factories.py` — readable parametrize that doesn't collapse under its own weight

4. `tasks/list` with Filtering and Pagination

5. Error Handling: `google.rpc.Status`