Top-Level Imports¶

Everything you need, one import away.

Flock exposes the most commonly used classes and utilities at the top level for convenient imports. No more hunting through submodules!

Quick Reference¶

from flock import (
    # Core
    Flock, flock_type, flock_tool, start_orchestrator,

    # Engines & Adapters
    DSPyEngine, BAMLAdapter, JSONAdapter, XMLAdapter, ChatAdapter, TwoStepAdapter,

    # Components (for extending)
    AgentComponent, EngineComponent, OrchestratorComponent, ServerComponent,
    AgentComponentConfig, OrchestratorComponentConfig, ServerComponentConfig,

    # Runtime (for custom engines/components)
    Context, EvalInputs, EvalResult,

    # Artifacts
    Artifact,

    # Visibility (access control)
    Visibility, PublicVisibility, PrivateVisibility, 
    LabelledVisibility, TenantVisibility, AfterVisibility, AgentIdentity,

    # Workflow control
    Until, When,

    # Advanced subscriptions
    BatchSpec, JoinSpec, ScheduleSpec,

    # Filtering
    FilterConfig,

    # Logging
    get_logger, configure_logging,
)

Categories¶

Core Orchestration¶

Import	Description	Learn More
`Flock`	Main orchestrator class	Quick Start
`flock_type`	Decorator to register Pydantic models as artifact types	Core Concepts
`flock_tool`	Decorator to register functions as agent tools	Agents Guide
`start_orchestrator`	Utility to start the orchestrator	Quick Start

from flock import Flock, flock_type

@flock_type
class Task(BaseModel):
    title: str
    priority: int

flock = Flock("openai/gpt-4.1")

Engines & Adapters¶

Import	Description	Learn More
`DSPyEngine`	Default DSPy-powered engine for LLM interactions	DSPy Engine Deep Dive
`BAMLAdapter`	BAML output format adapter	DSPy Engine - Adapters
`JSONAdapter`	JSON output format adapter	DSPy Engine - Adapters
`XMLAdapter`	XML output format adapter	DSPy Engine - Adapters
`ChatAdapter`	Chat-style output format adapter	DSPy Engine - Adapters
`TwoStepAdapter`	Two-step reasoning adapter	DSPy Engine - Adapters

from flock import Flock, DSPyEngine, BAMLAdapter

# Use a specific adapter
engine = DSPyEngine(
    model="openai/gpt-4.1",
    adapter=BAMLAdapter()
)

agent = (
    flock.agent("processor")
    .consumes(Input)
    .publishes(Output)
    .with_engines(engine)
)

DSPy Engine Deep Dive Local Models (Transformers)

Components¶

Import	Description	Learn More
`AgentComponent`	Base class for custom agent components	Agent Components Guide
`AgentComponentConfig`	Configuration for agent components	Agent Components Guide
`EngineComponent`	Base class for custom engines	Custom Engines Tutorial
`OrchestratorComponent`	Base class for orchestrator-level components	Orchestrator Components Guide
`OrchestratorComponentConfig`	Configuration for orchestrator components	Orchestrator Components Guide
`ServerComponent`	Base class for custom HTTP server components	Server Components Guide
`ServerComponentConfig`	Configuration for server components	Server Components Guide

from flock import AgentComponent, Context, EvalInputs, EvalResult

class LoggingComponent(AgentComponent):
    async def on_pre_evaluate(
        self, agent, ctx: Context, inputs: EvalInputs
    ) -> EvalInputs:
        print(f"Agent {agent.name} processing {len(inputs.artifacts)} artifacts")
        return inputs

Agent Components Orchestrator Components Server Components

Runtime Types¶

Import	Description	Learn More
`Context`	Execution context passed to components/engines	Custom Engines Tutorial
`EvalInputs`	Input wrapper containing artifacts and state	Custom Engines Tutorial
`EvalResult`	Result wrapper from engine evaluation	Custom Engines Tutorial

from flock import EngineComponent, Context, EvalInputs, EvalResult

class CustomEngine(EngineComponent):
    async def evaluate(
        self, agent, ctx: Context, inputs: EvalInputs, output_group
    ) -> EvalResult:
        # Access input artifacts
        for artifact in inputs.artifacts:
            print(f"Processing: {artifact.type}")

        # Return results
        return EvalResult(artifacts=[...])

Custom Engines Tutorial

Artifacts¶

Import	Description	Learn More
`Artifact`	Core artifact class for blackboard data	Blackboard Guide

from flock import Artifact

# Type hints for artifact handling
def process_artifact(artifact: Artifact) -> dict:
    return {
        "type": artifact.type,
        "producer": artifact.produced_by,
        "payload": artifact.payload
    }

Blackboard Guide

Visibility Controls¶

Import	Description	Learn More
`Visibility`	Base visibility class	Visibility Guide
`PublicVisibility`	Artifacts visible to all agents	Visibility - Public
`PrivateVisibility`	Artifacts visible only to specific agents	Visibility - Private
`LabelledVisibility`	Visibility based on agent labels	Visibility - Labelled
`TenantVisibility`	Multi-tenant visibility	Visibility - Tenant
`AfterVisibility`	Time-delayed visibility	Visibility - After
`AgentIdentity`	Agent identity for visibility checks	Visibility Guide

from flock import PrivateVisibility, TenantVisibility

# Private to specific agents
agent.publishes(
    SensitiveData, 
    visibility=PrivateVisibility(agents={"admin", "auditor"})
)

# Multi-tenant isolation
agent.publishes(
    CustomerData, 
    visibility=TenantVisibility(tenant_id="customer_123")
)

Visibility Controls Guide

Workflow Control¶

Import	Description	Learn More
`Until`	DSL for workflow termination conditions	Workflow Control Guide
`When`	DSL for subscription activation conditions	Workflow Control Guide

from flock import Until, When

# Stop when you have enough results
await flock.run_until(
    Until.artifact_count(Result).at_least(5),
    timeout=60
)

# Composite conditions
stop_condition = (
    Until.artifact_count(Analysis).at_least(3) |
    Until.workflow_error(correlation_id)
)
await flock.run_until(stop_condition, timeout=120)

# Conditional subscription activation
agent.consumes(
    Approval,
    when=When.correlation(Vote).count_at_least(3)
)

Workflow Control Guide

Subscription Patterns¶

Import	Description	Learn More
`BatchSpec`	Configure batch processing of artifacts	Batch Processing Guide
`JoinSpec`	Correlate related artifacts	Join Operations Guide
`ScheduleSpec`	Timer-based scheduling	Timer Scheduling Guide

from flock import BatchSpec, JoinSpec, ScheduleSpec
from datetime import timedelta

# Batch processing
agent.consumes(Task, batch=BatchSpec(size=10, timeout=5.0))

# Join related artifacts
agent.consumes(
    Order,
    join=JoinSpec(
        with_types=[Customer, Inventory],
        on="order_id"
    )
)

# Timer-based execution
agent.schedule(every=timedelta(minutes=5))

Batch Processing Join Operations Timer Scheduling

Filtering¶

Import	Description	Learn More
`FilterConfig`	Configuration for context/store filtering	Context Providers Guide

from flock import FilterConfig
from flock.core.context_provider import FilteredContextProvider

# Filter context by tags
provider = FilteredContextProvider(
    FilterConfig(tags={"urgent", "critical"}),
    limit=50
)
flock = Flock("openai/gpt-4.1", context_provider=provider)

Context Providers Guide

Logging¶

Import	Description	Learn More
`get_logger`	Get a Flock logger instance for your module	Distributed Tracing
`configure_logging`	Configure logging level and formatting	Configuration

from flock import get_logger, configure_logging
import logging

# Configure logging level
configure_logging(level=logging.DEBUG)

# Get a logger for your module
logger = get_logger(__name__)

logger.info("Starting workflow")
logger.debug("Processing artifact", extra={"artifact_id": "123"})
logger.warning("Retrying operation")

Distributed Tracing

Migration from Deep Imports¶

If you're using deep imports, here's how to migrate:

# Before (deep imports)
from flock.engines import DSPyEngine
from flock.components.agent import AgentComponent, EngineComponent
from flock.core.visibility import PrivateVisibility
from flock.utils.runtime import Context, EvalInputs, EvalResult

# After (top-level imports)
from flock import (
    DSPyEngine,
    AgentComponent, EngineComponent,
    PrivateVisibility,
    Context, EvalInputs, EvalResult,
)

Both styles work—use whichever you prefer. The deep imports are still available for cases where you need to import less common utilities.

Complete Import List¶

__all__ = [
    # Core
    "Flock",
    "flock_tool",
    "flock_type",
    "main",
    "start_orchestrator",
    # Engines
    "BAMLAdapter",
    "ChatAdapter",
    "DSPyEngine",
    "JSONAdapter",
    "TwoStepAdapter",
    "XMLAdapter",
    # Components
    "AgentComponent",
    "AgentComponentConfig",
    "EngineComponent",
    "OrchestratorComponent",
    "OrchestratorComponentConfig",
    "ServerComponent",
    "ServerComponentConfig",
    # Runtime
    "Context",
    "EvalInputs",
    "EvalResult",
    # Artifacts
    "Artifact",
    # Visibility
    "AfterVisibility",
    "AgentIdentity",
    "LabelledVisibility",
    "PrivateVisibility",
    "PublicVisibility",
    "TenantVisibility",
    "Visibility",
    # Conditions
    "Until",
    "When",
    # Subscriptions
    "BatchSpec",
    "JoinSpec",
    "ScheduleSpec",
    # Store
    "FilterConfig",
    # Logging
    "configure_logging",
    "get_logger",
]