We had a working network automation project (sfacts – simplefacts) with device discovery, fact collection, and NetBox IPAM integration. The CLI worked well for manual runs, but we wanted to add workflow orchestration, scheduling, and better observability for production use. We could create flashy UI with Streamlit of NiceGUI on top of existing API, they are both great. But why if there’s another great tool already. Prefect is the perfect fit!
In this guide, we will discuss how to integrate Prefect into our existing workflows. The challenge? Adding Prefect without rewriting too many lines of working code. My architectural choice was decoupling over granular observability – I will explain why.
I could have gone down the path of adding @task or @flow decorators everywhere and refactoring everything. But that felt wrong. Instead, I took advantage of our existing API-first design that already had clean separation between business logic and orchestration. Here’s how I added Prefect workflow orchestration without touching the core automation code.
┌─────────────────────────────────────────────────────────────────────────────────┐
│ CLI vs PREFECT INTERFACES │
└─────────────────────────────────────────────────────────────────────────────────┘
CLI (Direct Access) Prefect (Orchestrated)
┌──────────────────────┐ ┌──────────────────────┐
│ $ sfacts discover │ │ Prefect UI │
│ $ sfacts facts │ │ - Click "Run" │
│ $ sfacts netbox sync │ │ - Set parameters │
└──────────────────────┘ │ - View artifacts │
│ │ - Monitor logs │
│ │ - Schedule runs │
▼ └──────────────────────┘
Same Core APIs ◀────────────────────────────────┘
(No code duplication)Most automation scripts mix business logic with execution logic. This makes adding workflow orchestration painful because you have to refactor everything. The solution? Separate core functionality into reusable API classes first. Then Prefect (or any orchestrator) becomes a thin wrapper, not a rewrite.
However, this decoupling introduces a fundamental tradeoff: you lose granular observability within the orchestrator itself. When Prefect acts strictly as a thin wrapper, it becomes a high-level manager rather than a micromanager. It sees the final HTTP status code but becomes completely blind to the internal execution steps, variable states, and granular function-level failures inside your API. You are effectively trading deep, centralized workflow visibility for architectural flexibility, shifting the burden of internal debugging from the orchestrator over to your logging and tracing tools.
Before (Bad – Tightly Coupled):
# Tightly coupled - hard to add Prefect
def main():
# Discovery logic mixed with execution
for ip in subnet:
connect_ssh(ip)
run_commands()
parse_output()
save_to_netbox()
After (Good – API-First):
# Clean API separation
class NetworkDiscovery:
def discover_subnet(self, subnet: str) -> List[Dict]:
"""Pure business logic - no execution details"""
return self.scanner.scan(subnet)
class BasicFactCollector:
def collect_facts(self, devices: List[Dict]) -> Dict:
"""Reusable API - works with or without Prefect"""
return self._gather_facts(devices)
Luckily, our existing codebase already had this clean separation:
NetworkDiscovery class – Pure discovery API in sfacts/core/discovery.pyBasicFactCollector class – Pure collection API in sfacts/core/collector.pyNetBoxSyncManager class – Pure sync API in sfacts/core/netbox/device_sync.pyNo Prefect dependencies in core code. This made adding Prefect straightforward — I just wrapped the existing APIs.
By taking this approach, we are able to seamlessly integrate Prefect without disrupting the existing functionality. Here’s high-level visual architecture diagram
┌─────────────────────────────────────┐
│ Layer 3: Prefect Flows │ ← Orchestration
│ (deployments, scheduling, UI) │
├─────────────────────────────────────┤
│ Layer 2: Prefect Tasks │ ← Thin wrappers
│ (@task decorators, retry logic) │
├─────────────────────────────────────┤
│ Layer 1: Core API Classes │ ← Business logic
│ (NetworkDiscovery, Collector, │
│ NetBoxSync - NO Prefect deps) │
└─────────────────────────────────────┘
Our core APIs in sfacts/core/:
NetworkDiscovery – Handles subnet scanning and device identificationBasicFactCollector – Gathers facts using NAPALM + TextFSMNetBoxSyncManager – Manages device data synchronization to NetBoxThese were already pure Python classes, fully testable and CLI-ready. No orchestration logic mixed in.
I created thin @task decorators that wrap the existing core APIs:
# sfacts/tasks/discover.py
@task(name="discover-devices", retries=2, retry_delay_seconds=30)
def discover_devices(
subnet: str,
credentials: Dict[str, str],
timeout: int = 10,
max_workers: int = 20,
) -> List[Dict[str, Any]]:
"""
Discover network devices on a subnet via SSH port scanning and Netmiko SSHDetect.
Credentials are injected into each discovered device dict so the list can be
passed directly to the collect_facts task without further transformation.
Args:
subnet: Network subnet in CIDR notation (e.g., '192.168.1.0/24')
credentials: SSH credentials with 'username', 'password', and 'secret' keys
timeout: SSH timeout in seconds
max_workers: Maximum number of parallel scanning workers
Returns:
List of discovered device dicts with credentials embedded
"""
logger = get_run_logger()
logger.info("Starting device discovery on subnet %s", subnet)
discovery = NetworkDiscovery(
credentials=credentials,
timeout=timeout,
max_workers=max_workers,
)
discovered: List[Dict[str, Any]] = discovery.discover_subnet(subnet) # type: ignore[assignment]
for device in discovered:
device["credentials"] = credentials
logger.info("Discovered %d device(s) on %s", len(discovered), subnet)
return discovered
Pattern: Instantiate core API class → Call core method → Return result.
Compose tasks into workflows with error handling and artifact creation:
# sfacts/flows/modular_flows.py
@flow(name="discover-network", log_prints=False, on_failure=[])
async def discover_network_flow(
target: str,
username: Optional[str] = None,
password: Optional[str] = None,
secret: str = "",
use_secret_block: bool = True,
username_block_name: str = "lab-username",
password_block_name: str = "lab-password",
timeout: int = 10,
max_workers: int = 20,
) -> Dict[str, Any]:
"""
Discover network devices using auto-detected mode (subnet or tunnel).
Auto-detects discovery mode from target format:
- Subnet: '192.168.1.0/24' or '10.0.0.0/16'
- Tunnel: 'localhost:2101-2116' or '192.168.1.1:3000-3010'
Creates artifacts:
- 'devices-list': Visual table of discovered devices
- 'devices-json': JSON device list for next pipeline step
Args:
target: Network target (CIDR subnet or host:port-range)
username: SSH username (optional, uses secret block if not provided)
password: SSH password (optional, uses secret block if not provided)
secret: Enable secret for privileged mode (default: "")
use_secret_block: Use Prefect Secret blocks for credentials (default: True)
username_block_name: Name of username Secret block (default: "lab-username")
password_block_name: Name of password Secret block (default: "lab-password")
timeout: SSH timeout in seconds
max_workers: Maximum number of parallel workers
Returns:
Dict with 'devices' (list), 'discovery_mode' (str), 'target' (str)
"""
I started by identifying what needed orchestration and checking for clean API boundaries.
My Starting Point:
NetworkDiscovery API already existedBasicFactCollector API already existedNetBoxSyncManager API already existedResult: Because the APIs were already clean, I was ready for Prefect integration quickly.
I created one task per core API method with retry logic and structured logging.
What I Built:
# Prefect tasks created in sfacts/tasks/
- collect_facts() # Prefect tasks for device fact collection.
- discover_devices() # Prefect tasks for network device discovery.
- sync_to_netbox() # Prefect tasks for NetBox IPAM/DCIM synchronization.Code Pattern:
@task(name="task-name", retries=2, retry_delay_seconds=30)
def task_wrapper(params):
logger = get_run_logger()
logger.info("Starting task...")
api = CoreAPIClass(params)
result = api.core_method()
logger.info("Task complete")
return result
Next, I composed tasks into workflows with artifact creation for data passing.
Initial Flows (Monolithic – Not Great):
full_inventory_flow – discover → collect → sync (all in one)discovery_flow – discovery onlyProblem: This required running everything together. I couldn’t retry individual steps, which was annoying.
I refactored to an artifact-based data pipeline so workflow steps could run independently. Creating artifacts allowed me to pass data from one workflow into another. For every run I have an option to either input new data or use artifacts from previously ran flows.
Modular Flows with Artifacts:
@flow(name="discover-network")
async def discover_network_flow(target: str):
devices = discover_devices(subnet=target)
await create_markdown_artifact(
key="devices-json",
markdown=f"```json\n{json.dumps(devices)}\n```"
)
return devices
@flow(name="collect-facts")
async def collect_facts_flow(use_latest_discovery: bool = True):
if use_latest_discovery:
devices_json = await _get_latest_artifact_data("devices-json")
devices = json.loads(devices_json)
facts = collect_facts(devices=devices)
await create_artifact("facts-json", facts)
return facts
Why This Was Better:
Finally, I created deployment definitions and integrated secure credential management using Prefect’s Secret blocks. Please keep in mind that I used secret block for NetBox url as well, there are cases when you may desire to keep the URL private, but using variables should be sufficient.
Deployment Creation:
# sfacts/deployments.py
discover_network = discover_network_flow.to_deployment(
name="discover-network",
description=(
"Discover network devices with auto-detection of subnet or tunnel mode. "
"Subnet example: '192.168.1.0/24' | Tunnel example: 'localhost:2101-2116'. "
"Uses Prefect Secret blocks 'lab-username' and 'lab-password' by default. "
"Creates 'devices-json' artifact."
),
tags=["sfacts", "discovery", "modular"],
parameters={
"target": "192.168.121.0/24",
"username": None,
"password": None,
"secret": "",
"use_secret_block": True,
"username_block_name": "lab-username",
"password_block_name": "lab-password",
"timeout": 10,
"max_workers": 20,
},
)
The Problem: Monolithic flows require running everything together. Manual JSON copying between steps is error-prone and tedious.
The Solution: Prefect Artifacts
Prefect artifacts let you store data from one flow and retrieve it in another:
# Flow 1: Create artifact
await create_markdown_artifact(
key="devices-json",
markdown=f"```json\n{json.dumps(devices)}\n```",
description="Device list for fact collection"
)
# Flow 2: Retrieve artifact automatically
async def _get_latest_artifact_data(artifact_key: str):
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterKey
from prefect.client.schemas.sorting import ArtifactSort
async with get_client() as client:
artifacts = await client.read_artifacts(
artifact_filter=ArtifactFilter(
key=ArtifactFilterKey(any_=[artifact_key])
),
sort=ArtifactSort.CREATED_DESC,
limit=1,
)
return artifacts[0].data
Our Pipeline:
┌──────────────────────────────────────────────────────────────┐
│ SFACTS PREFECT ARTIFACT PIPELINE │
└──────────────────────────────────────────────────────────────┘
discover-network → collect-facts → sync-to-netbox
│ ▲ ▲
│ │ │
└──devices-json──────┘ │
│ │
└────facts-json────┘
Legend:
───────
Deployments run independently
Artifacts are auto-fetched as inputs for the next deploymentWhy This Works Well:
Real-World Usage:
discover-network deployment with target subnetdevices-json artifactcollect-facts with use_latest_discovery=Truefacts-json artifactsync-to-netbox with use_latest_facts=TrueThe Challenge: Network automation needs SSH credentials, NetBox needs API tokens. You obviously can’t hardcode these in your flows.
The Solution: Prefect Secret Blocks
# Create secrets (one-time setup)
from prefect.blocks.system import Secret
username_secret = Secret(value="admin")
await username_secret.save(name="lab-username")
password_secret = Secret(value="secret123")
await password_secret.save(name="lab-password")
# Use in flows
@flow(name="discover-network")
async def discover_network_flow(
target: str,
use_secret_block: bool = True,
username_block_name: str = "lab-username"
):
if use_secret_block:
username_secret = await Secret.load(username_block_name)
username = username_secret.get()
credentials = {"username": username, "password": password}
devices = discover_devices(subnet=target, credentials=credentials)
The Secret Blocks I Created:
lab-username – SSH usernamelab-password – SSH passwordnetbox-url – NetBox instance URL (doesn’t have to be a secret necessarily)netbox-api-token – NetBox API tokenWhy This Is Better:
Unit Tests for Tasks:
# tests/unit/test_tasks/test_discover_task.py
from unittest.mock import patch
from sfacts.tasks.discover import discover_devices
def test_discover_devices_task(mock_discovery):
with patch('sfacts.core.discovery.NetworkDiscovery') as mock:
mock.return_value.discover_subnet.return_value = [
{"ip": "192.168.1.1", "hostname": "router01"}
]
result = discover_devices(
subnet="192.168.1.0/24",
credentials={"username": "admin"}
)
assert len(result) == 1
assert result[0]["hostname"] == "router01"
Flow Tests with Prefect Test Harness:
# tests/unit/test_flows/test_discovery_flow.py
from prefect.testing.utilities import prefect_test_harness
def test_discovery_flow():
with prefect_test_harness():
result = discover_network_flow(target="192.168.1.0/24")
assert result["device_count"] > 0
# 1. Install with workflows extra
pip install sfacts[workflows]
# 2. Start Prefect server
prefect server start
# 3. Create Secret blocks (one-time setup)
python setup_secrets.py
# 4. Serve deployments
python serve.py
Available Deployments:
Production Workflow Example:
192.168.121.0/24, uses Secret blocks, creates devices-json artifactuse_latest_discovery: True, auto-fetches devices, creates facts-json artifactuse_latest_facts: True, bulk_sync: True (10–50x faster), syncs to NetBox IPAMScheduling:
discover_network = discover_network_flow.to_deployment(
name="discover-network",
schedule=CronSchedule(cron="0 2 * * *") # Daily at 2 AM
)
Monitoring:
By the Numbers:
What Actually Improved:
1. Observability
2. Flexibility
3. Collaboration
4. Maintainability
5. Future-Proofing
Critical Success Factors:
1. API-First Design is Non-Negotiable
2. Keep Tasks Thin
3. Use Artifacts for Data Passing
4. Leverage Secret Blocks
5. Test Core APIs, Mock Prefect
Common Pitfalls I Avoided (or Hit and Fixed):
Pitfall 1: Rewriting Everything
Symptom: Adding @task decorators directly to existing functions.
Fix: Create new wrapper tasks instead, leave core code alone.
Pitfall 2: Tight Coupling
Symptom: Core code imports Prefect.
Fix: Only tasks/flows import Prefect.
Pitfall 3: Monolithic Flows
Symptom: One giant flow that does everything.
Fix: Modular flows with artifact-based data passing.
Pitfall 4: Hardcoded Credentials
Symptom: Credentials in flow parameters.
Fix: Use Secret blocks.
When This Approach Makes Sense:
When This Approach Doesn’t Make Sense:
I added workflow orchestration to our network automation project in very little time without rewriting any of the core code. The key? The project already had API-first design with clean separation between business logic and orchestration. Of course there are tradeoffs: separation versus full observability, but that’s my current choice. As a next step we can create more complex deployments and dive deeper into automation with Prefect.
Your Action Plan:
What’s Next for This Project:
The best time to add workflow orchestration is when you don’t need to rewrite your code. If you build your automation with clean APIs from day one, adding orchestration later becomes a simple wrapper, not a complete rewrite.
