Prefect

Prefect uses Python decorators (@flow, @task) to transform standard functions into orchestrated units with built-in state management. The execution engine wraps decorated functions to automatically track execution state (Pending, Running, Completed, Failed, Cached) through a state machine, enabling recovery and observability without modifying core business logic. State transitions are persisted to the backend database and queryable via the Prefect Client.

Prefect's execution engine implements configurable retry logic at the task level using exponential backoff with jitter. When a task fails, the engine automatically re-executes it up to a specified retry count, with delays that grow exponentially (e.g., 1s, 2s, 4s, 8s). Retry policies are defined via @task decorators and stored in task metadata, allowing fine-grained control per task without modifying business logic.

Prefect exposes a REST API (FastAPI-based) for all operations: creating flows, submitting runs, querying logs, managing blocks, and configuring automations. The Python client (PrefectClient) wraps the REST API and provides a Pythonic interface for SDK users. The client handles authentication (API key-based), connection pooling, and automatic retries. Both API and client support async operations for high-throughput scenarios.

Prefect Server (self-hosted or Cloud) implements multi-tenancy with separate workspaces per tenant, role-based access control (RBAC) for flows/deployments/blocks, and audit logging of all API operations. The server uses FastAPI with SQLAlchemy ORM for database abstraction, supporting PostgreSQL and SQLite backends. Authentication is API key-based with scoped permissions (e.g., 'read flows', 'create deployments'). All operations are logged to the audit log with user, timestamp, and action metadata.

Prefect provides an MCP server that exposes Prefect operations (create flows, submit runs, query logs) as tools for AI models. The MCP server implements the Model Context Protocol, allowing Claude or other AI assistants to interact with Prefect via natural language. Users can ask the AI to 'create a flow that processes S3 files' and the AI generates Prefect code and submits it via MCP tools. The MCP server handles authentication and translates AI requests to Prefect API calls.

Prefect uses context variables (via Python's contextvars module) to inject runtime information into flows and tasks without explicit parameter passing. The context includes flow run ID, task run ID, logger, and custom variables. Parameters can be passed to flows at submission time and accessed via the context or function arguments. The system supports parameter validation via Pydantic models, enabling type-safe parameter handling.

Prefect provides task-level result caching that stores task outputs in a configurable cache backend (local filesystem, S3, or custom). Cache keys are generated from task name, version, and input parameters, allowing downstream tasks to skip execution if a cached result exists within the TTL. The cache is queryable and can be manually invalidated via the CLI or API.

Prefect's deployment system supports scheduling flows via cron expressions or fixed intervals (e.g., every 6 hours). Schedules are defined in deployment configuration and managed by the Prefect Server, which uses a background scheduler service to emit flow run events at scheduled times. Workers poll for scheduled runs and execute them in their configured work pools, with full observability into scheduled vs. ad-hoc runs.

Prefect's worker system enables distributed task execution across multiple machines or containers. Deployments are assigned to work pools (e.g., 'kubernetes', 'docker', 'local'), and workers poll the Prefect Server for flow runs assigned to their pool. Workers execute tasks with configurable concurrency limits (e.g., max 5 concurrent tasks per worker), and the server enforces global concurrency limits per work pool. Task execution can be containerized (Docker) or run directly on the worker machine.

Prefect's event system allows flows to be triggered by external events (e.g., S3 object creation, database updates) via custom automation rules. Events are emitted by integrations or custom code and stored in the Prefect Server's event log. Automation rules match events against patterns (e.g., 'event.resource.id matches "s3://my-bucket/*"') and trigger flow runs with parameterized inputs. Rules are evaluated server-side and can be created/updated via API or UI.

Prefect's logging system captures all task and flow execution logs and aggregates them in the Prefect Server database. Logs are structured (JSON-formatted with metadata like task name, run ID, timestamp) and queryable via the API and UI. The logging system integrates with Python's standard logging module, allowing custom loggers to emit logs that are automatically captured and stored. Logs can be streamed to external systems (e.g., Datadog, Splunk) via integrations.

Prefect's deployment system packages flow code and dependencies into versioned deployments that can be deployed to workers. Deployments are created via CLI (prefect deploy) or API and include flow code, Python dependencies (via pyproject.toml or requirements.txt), and configuration metadata. The system supports multiple deployment strategies: push deployments (code sent to worker at runtime) and pull deployments (worker pulls code from Git or S3). Deployments are versioned and can be rolled back via API.

Prefect's block system provides a declarative way to manage credentials and connections to external systems (databases, cloud storage, APIs) without hardcoding secrets. Blocks are configuration objects stored in the Prefect Server database and encrypted at rest. They can be referenced in flow code via block.load() and are injected at runtime. Blocks support templating and can be created via CLI, API, or UI. Common blocks include DatabaseBlock, S3Block, GCSBlock, and custom blocks can be defined by users.

Prefect's execution engine implements a state machine for flow runs with states (Scheduled, Pending, Running, Completed, Failed, Cancelled) and transitions. Tasks can conditionally execute based on upstream task states via if/else logic or the .result() method, enabling dynamic DAGs. The engine evaluates task dependencies at runtime, allowing downstream tasks to branch based on upstream results without pre-defining the full DAG structure.