Who is Muhammad Usman Akbar?

Muhammad Usman Akbar is a world-class AI Transformation Consultant and Agentic Architect focused on achieving 30x industrial efficiency through autonomous ecosystems.

What results can an AI Transformation Consultant provide?

By replacing manual work with autonomous AI workflows, a consultant like Muhammad Usman can deliver up to 30x growth in output while reducing operational overhead by 40%.

What is Agentic AI Orchestration?

It is the engineering of multi-agent systems where autonomous AI entities collaborate to manage complex industrial operations in production environments.

Managing Workflows

You've authored your first workflow with activities. It runs, processes tasks, and returns results. But what happens when you need to:

Check if a workflow is still running or has failed?
Wait for a manager's approval before proceeding with a high-value transaction?
Cancel a workflow that's stuck or no longer needed?
Clean up old workflow data that's consuming state store space?

Authoring workflows is only half the story. Production systems require workflow management: starting instances on demand, querying their status, injecting external events, and cleaning up completed runs. This lesson gives you the operational toolkit for workflow lifecycle management.

The Workflow Management API

Dapr's DaprWorkflowClient provides the control plane for your workflows. While WorkflowRuntime handles execution (registering and running workflows), DaprWorkflowClient handles management (starting, querying, and controlling workflow instances).

Think of it like this:

Component	Role	Analogy
WorkflowRuntime	Executes workflows	The factory floor
DaprWorkflowClient	Manages workflow instances	The control room

The control room operators don't assemble products; they monitor production lines, start new runs, and intervene when something goes wrong.

Starting Workflows

Using the Python SDK

The schedule_new_workflow method starts a new workflow instance:

python

from dapr.ext.workflow import DaprWorkflowClient

client = DaprWorkflowClient()

# Start a workflow with auto-generated instance ID
instance_id = client.schedule_new_workflow(
    workflow=task_processing_workflow,
    input={"task_id": "task-123", "title": "Review PR #456"}
)
print(f"Started workflow: {instance_id}")

Output:

text

Started workflow: abc123-def456-ghi789

You can also specify a custom instance ID for easier tracking:

python

# Start with custom instance ID
instance_id = client.schedule_new_workflow(
    workflow=task_processing_workflow,
    input={"task_id": "task-123", "title": "Review PR #456"},
    instance_id="task-123-workflow"
)
print(f"Started workflow: {instance_id}")

Custom instance IDs are valuable for Idempotency, Correlation, and Debugging.

Using the CLI

The Dapr CLI provides workflow management without writing code:

bash

# Start a workflow with input data
dapr workflow run task_processing_workflow \
  --app-id task-service \
  --input '{"task_id": "task-123", "title": "Review PR"}'

# Start with specific instance ID
dapr workflow run task_processing_workflow \
  --app-id task-service \
  --instance-id task-123-workflow \
  --input '{"task_id": "task-123", "title": "Review PR"}'

Querying Workflow Status

Once a workflow is running, you need visibility into its state. The get_workflow_state method returns comprehensive status information:

python

from dapr.ext.workflow import DaprWorkflowClient

client = DaprWorkflowClient()

# Query workflow status
state = client.get_workflow_state(
    instance_id="task-123-workflow",
    fetch_payloads=True  # Include input/output data
)

print(f"Instance ID: {state.instance_id}")
print(f"Status: {state.runtime_status}")
print(f"Created: {state.created_at}")

Runtime Status Values

Status	Meaning
RUNNING	Workflow is actively executing or waiting for an event
COMPLETED	Workflow finished successfully with output
FAILED	Workflow threw an unhandled exception
TERMINATED	Workflow was forcibly stopped via terminate_workflow
SUSPENDED	Workflow was paused via pause_workflow
PENDING	Workflow is scheduled but hasn't started yet

Raising External Events

External events enable human-in-the-loop workflows. A workflow pauses with wait_for_external_event, and an external process sends the event to resume execution.

Workflow Side: Waiting for Events

python

@wfr.workflow(name="approval_workflow")
def approval_workflow(ctx: wf.DaprWorkflowContext, order: dict):
    if order["amount"] > 1000:
        yield ctx.call_activity(request_approval, input=order)

        # Wait for external event OR timeout
        approval_event = ctx.wait_for_external_event("approval_decision")
        timeout = ctx.create_timer(timedelta(days=3))

        winner = yield wf.when_any([approval_event, timeout])

        if winner == timeout:
            return {"status": "rejected", "reason": "approval_timeout"}

        decision = approval_event.get_result()
        if not decision.get("approved"):
            return {"status": "rejected", "reason": decision.get("reason")}

    yield ctx.call_activity(process_order, input=order)
    return {"status": "completed"}

Raising Events via SDK

python

client.raise_workflow_event(
    instance_id="order-12345-workflow",
    event_name="approval_decision",
    data={"approved": True, "approver": "manager@company.com"}
)

Raising Events via CLI

bash

dapr workflow raise-event order-12345-workflow/approval_decision \
  --app-id order-service \
  --input '{"approved": true, "approver": "manager@company.com"}'

Pausing and Resuming Workflows

For administrative control, you can pause a running workflow and resume it later:

python

client.pause_workflow(instance_id="task-123-workflow")
client.resume_workflow(instance_id="task-123-workflow")

CLI Pause/Resume

bash

# Pause with reason
dapr workflow suspend task-123-workflow --app-id task-service

# Resume
dapr workflow resume task-123-workflow --app-id task-service

Terminating Workflows

Termination forcibly stops a workflow immediately. It is marked as TERMINATED.

python

client.terminate_workflow(instance_id="task-123-workflow")

Terminate vs Pause

Operation	Effect	Recoverable?	Use When
Pause	Suspends execution	Yes (resume)	Temporary hold, maintenance
Terminate	Stops execution permanently	No	Workflow is broken or unneeded

Warning: Termination does not run compensation logic.

Purging Workflow History

Completed workflows leave history in the state store. Purging removes it permanently.

python

# Purge a specific completed workflow
client.purge_workflow(instance_id="task-123-workflow")

CLI Purge Operations

bash

# Purge all completed workflows older than 30 days
dapr workflow purge --app-id task-service --all-older-than 720h

# Purge all terminal workflows
dapr workflow purge --app-id task-service --all

Complete Management Example

Here's a FastAPI application exposing workflow management endpoints:

python

from fastapi import FastAPI, HTTPException
from dapr.ext.workflow import WorkflowRuntime, DaprWorkflowClient, DaprWorkflowContext
from contextlib import asynccontextmanager

wfr = WorkflowRuntime()

@wfr.workflow(name="approval_workflow")
def approval_workflow(ctx: DaprWorkflowContext, order: dict):
    if order["amount"] > 1000:
        yield ctx.call_activity(request_approval_activity, input=order)
        approval = ctx.wait_for_external_event("approval_decision")
        decision = yield approval
        if not decision.get("approved"):
            return {"status": "rejected", "reason": decision.get("reason")}

    yield ctx.call_activity(process_order_activity, input=order)
    return {"status": "completed"}

@asynccontextmanager
async def lifespan(app: FastAPI):
    wfr.start()
    yield
    wfr.shutdown()

app = FastAPI(lifespan=lifespan)

@app.post("/orders/{order_id}/start")
async def start_order_workflow(order_id: str, amount: float = 500):
    client = DaprWorkflowClient()
    instance_id = client.schedule_new_workflow(
        workflow=approval_workflow,
        input={"order_id": order_id, "amount": amount},
        instance_id=f"order-{order_id}-workflow"
    )
    return {"instance_id": instance_id, "status": "started"}

@app.post("/orders/{order_id}/approve")
async def approve_order(order_id: str, approved: bool = True):
    client = DaprWorkflowClient()
    client.raise_workflow_event(
        instance_id=f"order-{order_id}-workflow",
        event_name="approval_decision",
        data={"approved": approved}
    )
    return {"event": "approval_decision", "sent": True}

Reflect on Your Skill

Does your dapr-deployment skill understand workflow management operations?

Test Your Skill

text

Using my dapr-deployment skill, explain how to implement a human approval step
in a Dapr workflow. Show both the workflow code that waits for approval and
the client code that sends the approval event.

If your skill covers wait_for_external_event and signal buffering, it's working correctly.

Try With AI

Prompt 1: Design an Approval Workflow

text

Help me design a purchase approval workflow with these requirements:
- Orders under $500: auto-approve
- Orders $500-$5000: requires manager approval within 3 days
- Orders over $5000: requires VP approval within 5 days
Show me the workflow code with wait_for_external_event and the FastAPI
endpoints for managers/VPs to submit their approval decisions.

Prompt 2: Build a Workflow Dashboard

text

I want to build a simple dashboard for monitoring my Dapr workflows. Help me create:
A FastAPI endpoint that lists all running workflows with their status

An endpoint that shows workflow history

Bulk operations: terminate all stuck workflows, purge all completed older than X days

Prompt 3: Event Pattern Troubleshooting

text

I'm having trouble with external events in my Dapr workflow. My workflow calls
wait_for_external_event("payment_confirmed") but even after I call
raise_workflow_event, the workflow doesn't continue. Help me debug this.

Safety Note: Workflow management operations are powerful. In production, implement access controls on management endpoints and log all operations for audit trails.