Let's build a small workflow that reads two regional request counts and combines them into one report. Along the way, we'll see how APFlow separates workflow structure from execution order and registers Python functions as capabilities that can be exposed to agents.
The example uses local sample data and SQLite. You can run it without a model provider or an API key.
APFlow is an open-source Python orchestration engine. It embeds the apcore module standard, giving registered capabilities descriptions and schemas that adapters can expose through MCP, A2A, REST, and CLI.
Use Python 3.11 or newer. Clone the repository and install APFlow in a virtual environment:
git clone https://github.com/aiperceivable/apflow.git
cd apflow
python -m venv .venv
source .venv/bin/activate
python -m pip install -e .
The activation command above is for a Unix-style shell.
Define the two relationships
All three tasks belong to one report:
Structure tree:
report
├── north
└── south
Execution dependencies:
north ──┐
├──> report
south ──┘
parent_id expresses membership in the report. dependencies makes the report wait for both regional results. A structural parent can therefore execute after its children when its dependencies require it.
The two regional tasks have no dependency on one another, so the engine can schedule them concurrently.
Register functions and run the graph
Save the following as workflow_demo.py in the cloned repository. The Pydantic models describe the capability inputs and business outputs. Functions receive input dictionaries; the decorator uses the models for their schemas.
import asyncio
import logging
from typing import Any
from pydantic import BaseModel, Field
from apflow.adapters.function_executor import function_executor
from apflow.app import create_app
class MetricInput(BaseModel):
region: str = Field(pattern="^(north|south)$")
class MetricOutput(BaseModel):
requests: int
class SummaryInput(BaseModel):
north: MetricOutput
south: MetricOutput
class SummaryOutput(BaseModel):
total_requests: int
@function_executor(
id="sample_metric",
description="Read a sample request count for one region",
input_model=MetricInput,
output_model=MetricOutput,
)
async def sample_metric(inputs: dict[str, Any]) -> dict[str, int]:
counts = {"north": 12, "south": 18}
return {"requests": counts[inputs["region"]]}
@function_executor(
id="summarize_metrics",
description="Combine request counts from north and south",
input_model=SummaryInput,
output_model=SummaryOutput,
)
async def summarize_metrics(inputs: dict[str, Any]) -> dict[str, int]:
total = inputs["north"]["requests"] + inputs["south"]["requests"]
return {"total_requests": total}
async def main() -> None:
app = create_app(connection_string="sqlite:///:memory:")
tasks: list[dict[str, Any]] = [
{
"id": "report",
"name": "Regional report",
"params": {"executor_id": "summarize_metrics"},
"dependencies": [{"id": "north"}, {"id": "south"}],
},
{
"id": "north",
"name": "North request count",
"parent_id": "report",
"params": {"executor_id": "sample_metric"},
"inputs": {"region": "north"},
},
{
"id": "south",
"name": "South request count",
"parent_id": "report",
"params": {"executor_id": "sample_metric"},
"inputs": {"region": "south"},
},
]
try:
tree = await app.task_creator.create_task_tree_from_array(tasks)
await app.task_manager.distribute_task_tree(tree)
app.session.expire_all()
report = await app.task_repository.get_task_by_id("report")
assert report is not None
assert report.status == "completed"
assert report.result == {"total_requests": 30}
logging.info("Report completed: %s", report.result)
finally:
app.session.close()
if name == "main":
logging.basicConfig(level=logging.INFO)
asyncio.run(main())
Run it:
python workflow_demo.py
Among the execution logs, you should see:
Report completed: {'total_requests': 30}
The assertions also verify the persisted status and business result. The example uses an in-memory database, so its records last for this process.
Follow the dependency data
APFlow injects each dependency's stored business output under its task ID. Here, summarize_metrics receives inputs["north"] and inputs["south"].
The function returns {"total_requests": 30} as business data. The function decorator handles the executor result envelope, and the task repository stores the business payload in task.result.
This distinction becomes useful when an executor also reports usage or a structured execution error. Those values have their own execution metadata rather than becoming fields that every downstream business function must unpack.
Inspect the capability contract
Because the decorators run before create_app(), the app's registry discovers the custom executors. Inside main(), you can inspect one definition:
definition = app.registry.get_definition("apflow.sample_metric")
assert definition is not None
logging.info("Input contract: %s", definition.input_schema)
For a server deployment, import your executor module before building the application's registry. A separate apflow serve --all process needs its own executor loading configuration; it does not inherit functions from this completed script.
You can explore APFlow's built-in HTTP interface independently:
apflow serve --all
In another terminal:
curl http://localhost:8000/modules
The unified server also provides /docs, /a2a, and /mcp. Custom modules loaded into that server become available through its configured adapters and policies.
Move from a local example to an operated workflow
This script demonstrates local execution, dependency injection, and capability registration. Background execution uses the separate durable submission contract: create definitions with an idempotency key, submit the root, and inspect persisted status.
REST and unified servers run a dispatcher. Standalone MCP, A2A, and CLI submission need one connected to the same database. For recovery, run.retry creates a new Run; submit its new root explicitly. Confirmed completed results can be reused while the original history remains available.
For a personal project, replace the sample metrics with one useful integration and keep the graph small. For a team, use a representative reporting or document-processing workflow to evaluate identity, failure recovery, and consumption reporting. Shared token scopes require participating executors to declare and enforce consumption bounds.
APFlow remains classified as Alpha. Start with a workflow whose outcomes you can verify, then add remote side effects and provider integrations deliberately.
The APFlow repository has the source and examples. See the current durable submission guide for the next step. If you try this example, open an issue with the integration you want to build, and star the repository to follow its development.
Top comments (0)