afd

AFD - Agent-First Development for Python

A Python toolkit for building applications with the Agent-First Development methodology.

Python follows the shared AFD feature set and agent-visible behavior, but keeps Pythonic APIs where that fits the language better. The goal is functional parity, not a byte-for-byte TypeScript port.

What is AFD?

Agent-First Development is a software development methodology where AI agents are treated as first-class users from day one. Instead of building UI first and adding API/agent access later, AFD inverts this:

Traditional:  UI → API → Agent Access (afterthought)
Agent-First:  Commands → Validation → UI (surface)

Core AFD surfaces stay framework-agnostic. React or browser integrations belong in examples and ecosystem layers, not in the afd core package.

Installation

# Core types only
pip install afd

# With MCP server support
pip install afd[server]

# With MCP client (network transports)
pip install afd[client]

# With CLI
pip install afd[cli]

# With testing utilities
pip install afd[testing]

# Everything
pip install afd[all]

Quick Start

Define a Command

from afd import CommandResult, success, error
from afd.server import define_command
from pydantic import BaseModel, Field

class Todo(BaseModel):
    id: str
    title: str
    done: bool = False

class CreateTodoInput(BaseModel):
    title: str = Field(..., min_length=1)

@define_command(
    name="todo-create",
    description="Create a new todo item",
    input_schema=CreateTodoInput,
)
async def create_todo(input: CreateTodoInput) -> CommandResult[Todo]:
    todo = Todo(id="todo-1", title=input.title)
    return success(
        data=todo,
        reasoning="Created new todo with default status",
    )

A handler receives the whole command input as its first argument. With input_schema, that input is validated and passed as a model instance; without it, the handler gets the raw input dict.

Create an MCP Server

from afd import ExposeOptions, success
from afd.server import create_server

server = create_server(
    name="todo-app",
    version="1.0.0",
)

# Todo and CreateTodoInput are defined in the example above.
@server.command(
    name="todo-create",
    description="Create a todo",
    input_schema=CreateTodoInput,
    expose=ExposeOptions(mcp=True),
)
async def create_todo(input: CreateTodoInput):
    todo = Todo(id="todo-1", title=input.title)
    return success(data=todo)

# Run the server (stdio for VS Code/Cursor).
# server.run(transport="sse") serves SSE at http://127.0.0.1:8000/sse instead.
server.run()

MCP exposure is opt-in. A command without expose=ExposeOptions(mcp=True) is invisible over MCP: it is not listed as a tool, and afd-call returns COMMAND_NOT_FOUND for it.

Test Your Commands

import pytest
from afd.testing import assert_success

# Use the mock_server fixture
async def test_create_todo(mock_server):
    @mock_server.command("todo-create")
    async def handler(input):
        from afd import success
        return success({"id": "1", "title": input["title"]})

    result = await mock_server.execute("todo-create", {"title": "Test"})

    data = assert_success(result)
    assert data["title"] == "Test"

Testing Helpers

Execute and validate commands with automatic timing and error wrapping:

from afd.testing import test_command, create_mock_command, create_test_context

# Run a handler with timing + validation
result = await test_command(my_handler, {"title": "Test"})
assert result.is_success
assert result.execution_time_ms >= 0

# Create mock commands for testing dependencies
cmd = create_mock_command("user-get", lambda inp: {"id": inp["id"]})
result = await cmd.handler({"id": 1}, None)
assert result.success

# Batch-test with expectations
from afd.testing import test_command_multiple

results = await test_command_multiple(my_handler, [
    {"input": {"title": "OK"}, "expect_success": True},
    {"input": {}, "expect_success": False, "expect_error": "VALIDATION_ERROR"},
])
assert all(r["passed"] for r in results)

Validators

Non-throwing validators return a ValidationResult for programmatic use:

from afd.testing import validate_result, validate_error, validate_command_definition

vr = validate_result(result)
assert vr.valid
assert len(vr.errors) == 0

# Validate with stricter options
from afd.testing import ResultValidationOptions
vr = validate_result(result, ResultValidationOptions(require_confidence=True))
for warning in vr.warnings:
    print(f"{warning.path}: {warning.message}")

Additional Assertions

from afd.testing import (
    assert_has_suggestion,   # Error includes recovery suggestion
    assert_retryable,        # Error retryable flag matches
    assert_step_status,      # Plan step has expected status
    assert_ai_result,        # Composite: confidence + reasoning + optional sources
)

# Validate error quality
error_result = error("NOT_FOUND", "Missing", suggestion="Check ID")
assert_has_suggestion(error_result)
assert_retryable(error_result, expected=False)

# Validate AI command output
ai_result = success(data, confidence=0.95, reasoning="Computed from input")
assert_ai_result(ai_result, min_confidence=0.9)

Core Types

CommandResult

The standard return type for all commands:

from afd import CommandResult, success, error

# Successful result
result = success(
    data={"id": "123"},
    reasoning="Created successfully",
    confidence=0.95,
)

# Error result
result = error(
    code="NOT_FOUND",
    message="Resource not found",
    suggestion="Check the ID and try again",
)

UX-Enabling Fields

AFD results include optional fields that enable rich agent experiences:

Wire Format

Results cross language boundaries as JSON in the same shape as the TypeScript and Rust packages (see spec/wire): keys are camelCase and unset fields are omitted, never null. Python attributes stay snake_case.

from afd import CommandResult, success
from afd.core.wire import to_wire

to_wire(success({"id": "1"}, undo_command="todo-delete"))
# {'success': True, 'data': {'id': '1'}, 'undoCommand': 'todo-delete'}

# Parsing accepts camelCase and snake_case
CommandResult.model_validate({"success": True, "undoCommand": "todo-delete"})

The MCP server sends failures with isError: true, and a missing or mistyped argument returns a VALIDATION_ERROR result with details.errors, details.missingFields and a suggestion. Every command result carries metadata.executionTimeMs and a per-call metadata.traceId, as in TypeScript. An exception from a handler or a middleware becomes a COMMAND_EXECUTION_ERROR whose message is "An internal error occurred" unless create_server(dev_mode=True). The core registry (create_command_registry), execute_pipeline and SimpleRegistry redact the same way and take the same dev_mode keyword.

Batches and Pipelines

afd-batch and afd-pipe run commands, never built-in tools: a batch item or pipeline step naming afd-batch, afd-pipe, afd-call, afd-discover or afd-detail rejects the whole request before anything runs, and so does a malformed when condition (for example $exist or {"$eq": "x"}).

• The envelope is checked as TypeScript checks it. A blank command name, null in place of an omitted optional field, or a string where a boolean or number belongs returns a failed BatchResult with INVALID_BATCH_REQUEST, or a PipelineResult whose one step (index -1) carries INVALID_PIPELINE_REQUEST, and nothing runs.
• A batch holds at most 500 commands, with parallelism at most 16. Configure the caps with create_server(max_batch_size=..., max_batch_parallelism=...); a larger request returns INVALID_BATCH_REQUEST.
• When a batch stops (a failure with stopOnError, or its timeout), commands still running are cancelled and awaited, so nothing keeps running after the result is sent. They report COMMAND_CANCELLED or BATCH_TIMEOUT.
• A handler that returns something other than a CommandResult gives its item or step an INVALID_COMMAND_RESULT failure instead of breaking the aggregate result.
• Pipelines run sequentially. options.parallel and a step with stream: true are not implemented: the offending step (step 0 for parallel) fails with UNSUPPORTED_OPTION and every other step is skipped, before any command runs. DirectClient.pipe() rejects a streaming step the same way.

Telemetry

Track command execution with standardized telemetry events:

from afd import create_telemetry_event, ConsoleTelemetrySink

# Create an event from execution data
event = create_telemetry_event(
    command_name="todo-create",
    started_at="2024-01-15T10:30:00.000Z",
    completed_at="2024-01-15T10:30:00.150Z",
    success=True,
    trace_id="trace-abc123",
)
# duration_ms is auto-calculated: 150.0

# Log to console (text or JSON format)
sink = ConsoleTelemetrySink(format="json")
sink.record(event)

# Implement a custom sink
class MyMonitoringSink:
    def record(self, event):
        send_to_monitoring(event.model_dump(exclude_none=True))

    def flush(self):
        pass

Middleware

Add cross-cutting concerns to command execution. As in TypeScript, middleware only runs for a call that reaches a handler: the server rejects an unknown, unexposed or out-of-context command, or invalid input, first. The input a middleware receives is the validated input the handler gets (the schema's model instance).

from afd.server import (
    create_server,
    default_middleware,
    compose_middleware,
    create_logging_middleware,
    create_timing_middleware,
    create_retry_middleware,
)

# Zero-config: logging, timing, and auto trace ID (returns a list)
server = create_server(name="todo-app", middleware=default_middleware())

# Or compose a custom stack into one middleware; the first one runs outermost
stack = compose_middleware(
    create_logging_middleware(),
    create_timing_middleware(slow_threshold=500),  # warn when a command takes over 500 ms
    create_retry_middleware(max_retries=3),
)
server = create_server(name="todo-app", middleware=[stack])

# Retries back off exponentially like TypeScript's createRetryMiddleware: retry n waits
# min(max_delay, retry_delay * 2 ** (n - 1)) ms (defaults 100 and 5000), randomized to
# between half and all of that unless jitter=False.
retry = create_retry_middleware(max_retries=5, retry_delay=200, max_delay=2000, jitter=False)

# Fixed-window rate limit: 100 calls per minute per client. Expired windows are
# evicted; at most max_keys (default 10,000) clients are tracked at once.
from afd.server import create_rate_limit_middleware

limit = create_rate_limit_middleware(
    max_requests=100,
    window_ms=60_000,
    key_fn=lambda ctx: ctx.extra.get("client_id", "anonymous"),
)

MCP Client (Network)

Connect to remote MCP servers over SSE or HTTP:

from afd import McpClient, McpClientConfig, create_client

# Quick setup
client = create_client("http://localhost:3100/sse")
await client.connect()

# Call a command (returns CommandResult)
result = await client.call("todo-create", {"title": "Hello"})
print(result.data)

# Raw tool call (no CommandResult wrapping)
raw = await client.call_tool("ping", {})

# Batch execution
batch_result = await client.batch([
    {"name": "todo-create", "input": {"title": "First"}},
    {"name": "todo-create", "input": {"title": "Second"}},
])

# Pipeline (variables: $prev, $first, $steps[N], $steps.<alias>, $input;
# see spec/pipeline-variables.md)
pipe_result = await client.pipe([
    {"command": "user-get", "input": {"id": 1}, "as": "user"},
    {"command": "order-list", "input": {"user_id": "$steps.user.id"}},
])

# Stream results
async for chunk in client.stream("long-task", {"query": "..."}):
    print(chunk)

await client.disconnect()

Use transports directly for lower-level control:

from afd.transports import SseTransport, HttpTransport, create_transport

transport = create_transport("sse", "http://localhost:3100/sse")
await transport.connect()
result = await transport.call_tool("ping", {})
await transport.disconnect()

Handoff Connections

Connect to streaming protocols (WebSocket, SSE) returned by handoff commands:

# Install with client dependencies
pip install afd[client]

from afd import (
    connect_handoff,
    create_reconnecting_handoff,
    register_builtin_handlers,
    HandoffConnectionOptions,
    ReconnectionOptions,
)
from afd.core.handoff import is_handoff

# Register built-in WebSocket and SSE handlers
register_builtin_handlers()

# Connect to a handoff result
result = await client.call('chat-connect', {'room_id': 'room-123'})

if result.success and is_handoff(result.data):
    # Simple connection
    conn = await connect_handoff(result.data, HandoffConnectionOptions(
        on_message=lambda msg: print('Message:', msg),
    ))

    # Or with auto-reconnect
    conn = await create_reconnecting_handoff(client, result.data,
        ReconnectionOptions(
            reconnect_command='chat-reconnect',
            session_id=result.data.get('credentials', {}).get('sessionId'),
            on_reconnect=lambda n: print(f'Reconnecting (attempt {n})'),
        ),
    )

    await conn.close()  # stops any reconnect in progress; nothing stays open

• The built-in handlers read in a background task: every message reaches on_message (parsed as JSON when it is JSON) and a close by the server reaches on_disconnect.
• A credentials.token is sent as an Authorization: Bearer header, never in the URL.
• A reconnecting connection retries with exponential backoff until it connects, runs out of attempts (on_reconnect_failed, state failed) or is closed. Before each attempt it calls reconnect_command with reconnect_args plus sessionId (the wire name, as in TypeScript) set from the session_id option.
• create_handoff() returns the wire format (sessionId, expectedLatency, expiresAt, maxAttempts, backoffMs); the handoff helpers accept camelCase or snake_case.

CLI

afd connect http://localhost:3100/sse   # SSE endpoint (an .../message URL uses HTTP)
afd tools
afd call todo-create '{"title": "Hello"}'

afd connect takes a server URL (or mock for testing); it does not start or name servers.

Packages

Related

• AFD TypeScript - Original TypeScript implementation
• AFD Philosophy - Why AFD?
• Command Schema Guide - Designing commands

License

MIT