Your First Workflow¶

Learn how to define and execute your first workflow in Adaptive Sentience.

What is a Workflow?¶

A workflow is a sequence of tool executions orchestrated across edge nodes. Workflows can:

Execute tools on specific nodes or node types
Handle offline scenarios with store-and-forward
Enforce capability-based policies
Provide execution evidence and audit trails

Simple Workflow: PII Redaction¶

Let's create a workflow that redacts personally identifiable information (PII) from text.

Step 1: Start Your Environment¶

Make sure your gateway and edge node are running:

# Terminal 1: Gateway
python -m gateway.http_gateway --host 127.0.0.1 --port 8787 --dev-token

# Terminal 2: Edge Node
cd edge_node
NODE_PORT=8000 python3 node.py

Step 2: Create Your Workflow Script¶

Create a file called my_first_workflow.py:

#!/usr/bin/env python3
import requests
import json

GATEWAY_URL = "http://127.0.0.1:8787"

def redact_pii(text):
    """Redact PII from text using the gateway."""

    # Define the workflow request
    request = {
        "target": {"kind": "local"},  # Target any local node
        "tool_name": "pii_redact",
        "tool_args": {"text": text},
        "token": {
            "capabilities": ["tool:pii_redact"]
        }
    }

    # Execute the workflow
    response = requests.post(
        f"{GATEWAY_URL}/v1/tool_call",
        json=request
    )

    return response.json()

if __name__ == "__main__":
    # Test data with PII
    sample_text = """
    Contact John Smith at john.smith@example.com or call 555-123-4567.
    His SSN is 123-45-6789.
    """

    print("Original text:")
    print(sample_text)
    print("\n" + "="*60 + "\n")

    # Execute workflow
    result = redact_pii(sample_text)

    if result.get("ok"):
        print("Redacted text:")
        print(result["result"]["redacted_text"])
        print("\nRedactions found:")
        for redaction in result["result"]["redactions"]:
            print(f"  - {redaction['type']}: positions {redaction['start']}-{redaction['end']}")

        print("\nExecution metadata:")
        print(f"  - Verified: {result['verified']}")
        print(f"  - Execution path: {result['execution_path']}")
        print(f"  - Degraded: {result['degraded']}")
    else:
        print("Error:", result.get("error"))

Step 3: Run the Workflow¶

python3 my_first_workflow.py

Expected output:

Original text:
    Contact John Smith at john.smith@example.com or call 555-123-4567.
    His SSN is 123-45-6789.

============================================================

Redacted text:
    Contact John Smith at [REDACTED] or call [REDACTED].
    His SSN is [REDACTED].

Redactions found:
  - email: positions 27-50
  - phone: positions 59-71
  - ssn: positions 85-97

Execution metadata:
  - Verified: True
  - Execution path: ['local:abc123']
  - Degraded: False

Understanding the Workflow¶

Request Structure¶

{
    "target": {"kind": "local"},      # Where to execute
    "tool_name": "pii_redact",         # Which tool to call
    "tool_args": {"text": "..."},      # Tool inputs
    "token": {                         # Authorization
        "capabilities": ["tool:pii_redact"]
    }
}

Response Structure¶

{
    "ok": True,                        # Success indicator
    "verified": True,                  # Cryptographic verification
    "result": {...},                   # Tool output
    "execution_path": ["local:abc123"], # Which nodes executed
    "degraded": False                  # Was quality degraded?
}

Multi-Step Workflow¶

Now let's create a workflow with multiple steps.

Create multi_step_workflow.py:

#!/usr/bin/env python3
import requests

GATEWAY_URL = "http://127.0.0.1:8787"

def call_tool(tool_name, tool_args, capabilities):
    """Generic tool call helper."""
    response = requests.post(
        f"{GATEWAY_URL}/v1/tool_call",
        json={
            "target": {"kind": "local"},
            "tool_name": tool_name,
            "tool_args": tool_args,
            "token": {"capabilities": capabilities}
        }
    )
    return response.json()

def secure_document_workflow(document_text):
    """Multi-step workflow: redact PII, then summarize."""

    print("Step 1: Redacting PII...")
    redaction_result = call_tool(
        "pii_redact",
        {"text": document_text},
        ["tool:pii_redact"]
    )

    if not redaction_result.get("ok"):
        return {"error": "Redaction failed"}

    redacted_text = redaction_result["result"]["redacted_text"]
    print(f"  ✓ Redacted {len(redaction_result['result']['redactions'])} items\n")

    print("Step 2: Summarizing document...")
    summary_result = call_tool(
        "summarize",
        {"text": redacted_text, "max_length": 100},
        ["tool:summarize"]
    )

    if not summary_result.get("ok"):
        return {"error": "Summarization failed"}

    print(f"  ✓ Generated summary\n")

    return {
        "redacted_text": redacted_text,
        "summary": summary_result["result"]["summary"],
        "execution_evidence": {
            "redaction_path": redaction_result["execution_path"],
            "summary_path": summary_result["execution_path"]
        }
    }

if __name__ == "__main__":
    document = """
    Incident Report - 2024-01-15

    Officer Jane Doe (badge #12345) responded to a disturbance at
    123 Main Street. Contact was made with resident John Smith
    (DOB: 1985-05-20, SSN: 987-65-4321).

    The resident reported suspicious activity around 3:00 PM.
    No evidence of criminal activity was found. Case closed.

    Officer contact: jane.doe@police.gov, 555-987-6543
    """

    result = secure_document_workflow(document)

    if "error" not in result:
        print("="*60)
        print("WORKFLOW COMPLETE")
        print("="*60)
        print("\nSummary:")
        print(result["summary"])
        print("\nExecution Evidence:")
        print(f"  Redaction executed on: {result['execution_evidence']['redaction_path']}")
        print(f"  Summary executed on: {result['execution_evidence']['summary_path']}")

Run it:

python3 multi_step_workflow.py

Targeting Specific Nodes¶

By Node Type¶

# Execute on mobile devices only
request = {
    "target": {"kind": "mobile"},  # android, ios
    "tool_name": "capture_photo",
    "tool_args": {},
    "token": {"capabilities": ["tool:capture_photo"]}
}

By Node ID¶

# Execute on a specific node
request = {
    "target": {"node_id": "local:abc123"},
    "tool_name": "get_location",
    "tool_args": {},
    "token": {"capabilities": ["tool:get_location"]}
}

By Capability¶

# Execute on any node with GPS capability
request = {
    "target": {"required_capabilities": ["gps"]},
    "tool_name": "get_location",
    "tool_args": {},
    "token": {"capabilities": ["tool:get_location"]}
}

Error Handling¶

Always check for errors in your workflows:

result = call_tool("some_tool", {...}, [...])

if not result.get("ok"):
    error = result.get("error", "Unknown error")

    if "permission" in error.lower():
        print("Permission denied - check capability token")
    elif "not found" in error.lower():
        print("Tool or node not found")
    elif "timeout" in error.lower():
        print("Request timed out - node may be offline")
    else:
        print(f"Error: {error}")
else:
    # Process successful result
    print(result["result"])

Testing Offline Operation¶

Run the workflow script
While it's running, stop the edge node (Ctrl+C)
The gateway will queue the request
Restart the edge node
The request will be delivered automatically

# Terminal 3: Run workflow
python3 my_first_workflow.py &

# Terminal 2: Stop node
# Press Ctrl+C

# Wait a few seconds...

# Terminal 2: Restart node
NODE_PORT=8000 python3 node.py

# Terminal 3: Check results
# The workflow should complete successfully!

Next Steps¶

Core Concepts - Deep dive into architecture
Tool Contracts - Create custom tools
Policy Enforcement - Add governance
API Reference - Full API documentation