Adding Orchestrator Workflow

Adding a New Orchestrator Workflow

This skill provides a comprehensive guide for adding new workflow endpoints to the QuAIA™ orchestrator. Workflows are FastAPI endpoints that trigger and coordinate agent tasks.

Overview

The orchestrator (orchestrator/main.py) exposes HTTP endpoints that:

1. Receive external requests (webhooks, API calls)
2. Route tasks to appropriate agents
3. Coordinate multi-agent workflows
4. Handle results and trigger follow-up actions

Workflow Architecture

A typical orchestrator workflow:

1. Receives a request (POST/GET endpoint)
2. Extracts relevant data from the request
3. Sends task(s) to agent(s) using _send_task_to_agent()
4. Parses the agent response using helper functions
5. Optionally triggers follow-up workflows
6. Returns the result to the caller

Step-by-Step Instructions

Step 1: Define the Request Model (if needed)

If your endpoint accepts structured input, create a request model in common/models.py:

📄 Template: resources/models_template.py

Step 2: Define the Response Model (if needed)

If the workflow returns structured data beyond simple status messages, add a response model (also in the template above).

Step 3: Create the Endpoint Function

Add your endpoint in orchestrator/main.py:

📄 Template: resources/endpoint_template.py

Step 4: Helper Functions Reference

The orchestrator provides these helper functions for working with agent tasks:

Sending Tasks to Agents

# Send a text task to an automatically selected agent
completed_task = await _send_task_to_agent(
    task_content: str,      # The payload/content for the agent
    task_description: str   # Used to select the appropriate agent
) -> Task

# Send a message with file attachments
completed_task = await _send_task_to_agent_with_message(
    message: Message,       # A2A Message with text and file parts
    task_description: str
) -> Task

Parsing Agent Responses

# Validate task completed successfully
_validate_task_status(task: Task, task_description: str)

# Extract artifacts from task (raises exception if none)
artifacts = _get_artifacts_from_task(task: Task, task_description: str) -> list[Artifact]

# Extract text content from artifacts
text_parts = _get_text_content_from_artifacts(
    artifacts: list[Artifact],
    task_description: str,
    any_content_expected: bool = True  # Set False if empty is OK
) -> list[str]

# Parse artifacts as a Pydantic model (also handles AgentExecutionError)
result = _get_model_from_artifacts(
    artifacts: list[Artifact],
    task_description: str,
    model_type: type[T]
) -> T | AgentExecutionError | None

# Extract file artifacts (screenshots, logs, etc.)
files = _get_file_contents_from_artifacts(artifacts: list[Artifact]) -> list[FileWithBytes]

Error Handling

# Raise HTTPException with consistent error handling and logging
_handle_exception(error_message: str, status_code: int = 500)

Step 5: Multi-Agent Workflows

For workflows that involve multiple agents in sequence:

📄 Example: examples/multi_agent_workflow.py

Step 6: Parallel Agent Execution

For workflows that can process items in parallel:

📄 Example: examples/parallel_execution.py

Step 7: Using Execution Lock (Optional)

For workflows that should not run concurrently (e.g., test execution):

@orchestrator_app.post("/exclusive-workflow")
async def exclusive_workflow(request: Request, api_key: str = Depends(_validate_api_key)):
    """Workflow that requires exclusive access."""
    
    async with execution_lock:  # Only one instance runs at a time
        # ... workflow logic ...
        return {"message": "Exclusive workflow completed"}

Step 8: Add Webhook URL Configuration (Optional)

If the endpoint will be called via webhooks, add the URL to config.py:

# Webhook URLs
<WORKFLOW_NAME>_WEBHOOK_URL = f"{ORCHESTRATOR_URL}/<endpoint-path>"

Step 9: Update README Documentation

Add documentation for the new endpoint in README.md under "Invoking Orchestrator Workflows":

### <Workflow Name>

Description of what this workflow does.

* **Endpoint:** `POST /<endpoint-path>`
  
  Example payload:
  ```json
  {
      "field_name": "value"
  }

Response:

{
    "message": "Workflow completed successfully",
    "result": { ... }
}

### Step 10: Create Unit Tests

Create test cases in `tests/orchestrator/test_endpoints.py` or a new file:

📄 **Example:** [examples/test_endpoint_example.py](examples/test_endpoint_example.py)

## Complete Workflow Example

For a full example including models and endpoint:

📄 **Example:** [examples/complete_workflow.py](examples/complete_workflow.py)

## Verification Checklist

After adding the workflow, verify:

- [ ] Request model (if any) added to `common/models.py`
- [ ] Response model (if any) added to `common/models.py`
- [ ] Endpoint function follows the standard pattern
- [ ] Proper error handling with `_handle_exception()`
- [ ] API key validation via `Depends(_validate_api_key)`
- [ ] Logging at key points (start, completion, errors)
- [ ] Unit tests cover success and failure cases
- [ ] Documentation updated in README.md
- [ ] Tests pass: `pytest tests/orchestrator/ -v`
- [ ] Endpoint accessible: `curl -X POST http://localhost:8000/<endpoint-path> -d '...'`

---
> Converted and distributed by [TomeVault](https://tomevault.io/claim/partarstu) — claim your Tome and manage your conversions.
<!-- tomevault:4.0:skill_md:2026-04-11 -->