External Query API Reference

External Query API Reference

Overview

The External Query API is a lightweight HTTP REST API for querying task and stream state. It enables external scripts (e.g., Python orchestration tools) to monitor progress without requiring MCP protocol integration.

Purpose

• External Integration: Enable non-MCP tools to monitor task state
• Orchestration: Support parallel stream management via external scripts
• Real-time Monitoring: Query current agent activity and task progress
• Dependency Polling: Check stream completion for dependency resolution

Architecture

• Embedded Server: Lightweight HTTP server process
• Framework: Fastify HTTP server (lightweight, production-ready)
• Protocol: Standard HTTP/REST (no authentication, localhost-only)
• Default URL: http://localhost:9090

Security Model

The API is designed for localhost-only access:

• Default bind address: 127.0.0.1 (not accessible from network)
• No authentication required (assumes trusted localhost environment)
• Read-only operations (no mutations)

Configuration

Configure the HTTP API via environment variables:

export HTTP_API_HOST="127.0.0.1"
export HTTP_API_PORT="9090"

Environment Variables

Server Lifecycle

The HTTP API server:

• Starts when the query API process starts
• Fails gracefully if port is in use
• Logs startup message: HTTP API listening on http://127.0.0.1:9090

Endpoints

GET /health

Health check endpoint for monitoring server availability.

Query Parameters: None

Response:

{
  "status": "ok",
  "timestamp": "2026-01-08T12:34:56.789Z"
}

Status Codes:

• 200 OK: Server is running

GET /api/streams

List all independent work streams with progress summary.

Query Parameters:

Response:

{
  "streams": [
    {
      "streamId": "Stream-A",
      "streamName": "foundation",
      "streamPhase": "foundation",
      "totalTasks": 4,
      "completedTasks": 4,
      "inProgressTasks": 0,
      "blockedTasks": 0,
      "pendingTasks": 0,
      "progressPercentage": 100,
      "files": [
        "src/types.ts",
        "src/tools/stream.ts"
      ],
      "dependencies": []
    },
    {
      "streamId": "Stream-B",
      "streamName": "command-updates",
      "streamPhase": "parallel",
      "totalTasks": 3,
      "completedTasks": 1,
      "inProgressTasks": 1,
      "blockedTasks": 0,
      "pendingTasks": 1,
      "progressPercentage": 33,
      "files": [
        ".claude/commands/protocol.md",
        ".claude/commands/continue.md"
      ],
      "dependencies": ["Stream-A"]
    }
  ],
  "totalStreams": 2
}

Status Codes:

• 200 OK: Success

Stream Fields:

GET /api/streams/:streamId

Get detailed information for a specific stream, including all tasks.

Path Parameters:

Query Parameters:

Response:

{
  "streamId": "Stream-B",
  "streamName": "command-updates",
  "streamPhase": "parallel",
  "dependencies": ["Stream-A"],
  "progressPercentage": 33,
  "tasks": [
    {
      "id": "TASK-456",
      "title": "Update /protocol command",
      "description": "Add stream routing logic",
      "status": "in_progress",
      "assignedAgent": "me",
      "prdId": "PRD-123",
      "parentId": null,
      "blockedReason": null,
      "notes": "Working on implementation",
      "metadata": {
        "streamId": "Stream-B",
        "streamName": "command-updates",
        "streamPhase": "parallel",
        "files": [".claude/commands/protocol.md"],
        "streamDependencies": ["Stream-A"],
        "complexity": "medium"
      },
      "createdAt": "2026-01-08T10:00:00.000Z",
      "updatedAt": "2026-01-08T11:30:00.000Z"
    },
    {
      "id": "TASK-457",
      "title": "Update /continue command",
      "description": "Add stream argument support",
      "status": "pending",
      "assignedAgent": "me",
      "prdId": "PRD-123",
      "parentId": null,
      "blockedReason": null,
      "notes": null,
      "metadata": {
        "streamId": "Stream-B",
        "streamName": "command-updates",
        "streamPhase": "parallel",
        "files": [".claude/commands/continue.md"],
        "streamDependencies": ["Stream-A"]
      },
      "createdAt": "2026-01-08T10:00:00.000Z",
      "updatedAt": "2026-01-08T10:00:00.000Z"
    }
  ]
}

Error Response (404 Not Found):

{
  "error": "Stream not found",
  "streamId": "Stream-X"
}

Status Codes:

• 200 OK: Stream found
• 404 Not Found: Stream does not exist or is archived (unless includeArchived=true)

Task Fields:

GET /api/tasks

Query tasks with flexible filtering.

Query Parameters:

Response:

{
  "tasks": [
    {
      "id": "TASK-456",
      "title": "Update /protocol command",
      "description": "Add stream routing logic",
      "status": "in_progress",
      "assignedAgent": "me",
      "prdId": "PRD-123",
      "parentId": null,
      "blockedReason": null,
      "notes": "Working on implementation",
      "metadata": {
        "streamId": "Stream-B",
        "streamName": "command-updates",
        "streamPhase": "parallel",
        "files": [".claude/commands/protocol.md"],
        "streamDependencies": ["Stream-A"],
        "complexity": "medium"
      },
      "createdAt": "2026-01-08T10:00:00.000Z",
      "updatedAt": "2026-01-08T11:30:00.000Z"
    }
  ],
  "totalTasks": 1
}

Status Codes:

• 200 OK: Success (empty array if no matches)

Notes:

• Archived tasks are excluded by default
• Tasks are ordered by created_at DESC (newest first)
• Combine filters for precise queries (e.g., status=in_progress&streamId=Stream-B)

GET /api/activity

Get current agent activity across all streams or filtered by stream.

Query Parameters:

Response:

{
  "activities": [
    {
      "streamId": "Stream-B",
      "streamName": "command-updates",
      "agentId": "me",
      "agentName": "Engineer",
      "taskId": "TASK-456",
      "taskTitle": "Update /protocol command",
      "activityDescription": "Implementing stream routing logic",
      "phase": "implementation",
      "startedAt": "2026-01-08T11:00:00.000Z",
      "lastHeartbeat": "2026-01-08T11:28:00.000Z",
      "isActive": true
    },
    {
      "streamId": "Stream-C",
      "streamName": "documentation",
      "agentId": "doc",
      "agentName": "Documentation",
      "taskId": "TASK-789",
      "taskTitle": "Write API reference",
      "activityDescription": null,
      "phase": null,
      "startedAt": "2026-01-08T09:00:00.000Z",
      "lastHeartbeat": "2026-01-08T10:45:00.000Z",
      "isActive": false
    }
  ],
  "totalActive": 1,
  "totalIdle": 1
}

Status Codes:

• 200 OK: Success

Activity Fields:

Agent Name Mappings:

Active Threshold:

• Agent is considered active if last heartbeat is within 5 minutes
• Idle agents have heartbeats older than 5 minutes

Usage Examples

Python Client Examples

List All Streams

import requests

def list_streams():
    """Get all active streams with progress."""
    response = requests.get('http://localhost:9090/api/streams')
    response.raise_for_status()
    data = response.json()

    print(f"Total streams: {data['totalStreams']}")
    for stream in data['streams']:
        print(f"{stream['streamId']}: {stream['streamName']} - {stream['progressPercentage']}% complete")
        print(f"  Tasks: {stream['completedTasks']}/{stream['totalTasks']}")
        print(f"  Phase: {stream['streamPhase']}")
        if stream['dependencies']:
            print(f"  Depends on: {', '.join(stream['dependencies'])}")

if __name__ == '__main__':
    list_streams()

Example Output:

Total streams: 2
Stream-A: foundation - 100% complete
  Tasks: 4/4
  Phase: foundation
Stream-B: command-updates - 33% complete
  Tasks: 1/3
  Phase: parallel
  Depends on: Stream-A

Check Stream Completion (Dependency Polling)

import requests
import time

def wait_for_stream_completion(stream_id, poll_interval=5, timeout=300):
    """Poll stream until completion or timeout."""
    start_time = time.time()

    while True:
        if time.time() - start_time > timeout:
            raise TimeoutError(f"Stream {stream_id} did not complete within {timeout}s")

        response = requests.get(f'http://localhost:9090/api/streams/{stream_id}')

        if response.status_code == 404:
            raise ValueError(f"Stream {stream_id} not found")

        response.raise_for_status()
        stream = response.json()

        completed = stream['completedTasks']
        total = len(stream['tasks'])
        progress = stream['progressPercentage']

        print(f"{stream_id}: {completed}/{total} tasks complete ({progress}%)")

        if completed == total:
            print(f"Stream {stream_id} completed!")
            return True

        # Check for blocked tasks
        blocked = sum(1 for t in stream['tasks'] if t['status'] == 'blocked')
        if blocked > 0:
            print(f"Warning: {blocked} blocked task(s)")

        time.sleep(poll_interval)

if __name__ == '__main__':
    # Wait for foundation stream before starting parallel work
    wait_for_stream_completion('Stream-A')
    print("Foundation complete, can start parallel streams")

Example Output:

Stream-A: 2/4 tasks complete (50%)
Stream-A: 3/4 tasks complete (75%)
Stream-A: 4/4 tasks complete (100%)
Stream-A completed!
Foundation complete, can start parallel streams

Monitor Agent Activity

import requests

def get_active_agents():
    """Get currently active agents."""
    response = requests.get('http://localhost:9090/api/activity?active=true')
    response.raise_for_status()
    data = response.json()

    if data['totalActive'] == 0:
        print("No active agents")
        return

    print(f"Active agents: {data['totalActive']}")
    for activity in data['activities']:
        if activity['isActive']:
            print(f"\n{activity['agentName']} ({activity['agentId']})")
            print(f"  Stream: {activity['streamName']}")
            print(f"  Task: {activity['taskTitle']}")
            if activity['activityDescription']:
                print(f"  Activity: {activity['activityDescription']}")
            if activity['phase']:
                print(f"  Phase: {activity['phase']}")

if __name__ == '__main__':
    get_active_agents()

Example Output:

Active agents: 2

Engineer (me)
  Stream: command-updates
  Task: Update /protocol command
  Activity: Implementing stream routing logic
  Phase: implementation

Documentation (doc)
  Stream: api-docs
  Task: Write API reference
  Phase: documentation

Query In-Progress Tasks

import requests

def get_in_progress_tasks():
    """Get all in-progress tasks."""
    response = requests.get('http://localhost:9090/api/tasks?status=in_progress')
    response.raise_for_status()
    data = response.json()

    print(f"In-progress tasks: {data['totalTasks']}")
    for task in data['tasks']:
        stream_name = task['metadata'].get('streamName', 'Unknown')
        agent = task.get('assignedAgent', 'Unassigned')
        print(f"\n{task['id']}: {task['title']}")
        print(f"  Stream: {stream_name}")
        print(f"  Agent: {agent}")
        if task['notes']:
            print(f"  Notes: {task['notes']}")

if __name__ == '__main__':
    get_in_progress_tasks()

Error Handling

Server Not Running

If the HTTP API server is not running, requests will fail with a connection error:

import requests

try:
    response = requests.get('http://localhost:9090/api/streams')
    response.raise_for_status()
except requests.exceptions.ConnectionError:
    print("Error: HTTP API is not running")
    print("Ensure the query API server is started")
except requests.exceptions.RequestException as e:
    print(f"HTTP error: {e}")

404 Stream Not Found

response = requests.get('http://localhost:9090/api/streams/Stream-X')

if response.status_code == 404:
    error = response.json()
    print(f"Error: {error['error']}")
    print(f"Stream ID: {error['streamId']}")
else:
    stream = response.json()
    # Process stream data

Port Already in Use

If the configured port is already in use, the HTTP API server will fail to start. Check logs:

Failed to start HTTP API server: Error: listen EADDRINUSE: address already in use 127.0.0.1:9090

Solution: Change HTTP_API_PORT environment variable to an available port.

Limitations

Read-Only Access

The External Query API provides read-only access to task state. Mutations (creating tasks, updating status, etc.) must be done via the tc CLI.

Why: Maintains single source of truth through the tc CLI with proper validation and activity logging.

No Authentication

The API does not implement authentication. It is designed for localhost-only access in trusted environments.

Security Model:

• Bind to 127.0.0.1 (not accessible from network)
• Assumes localhost is trusted
• No sensitive data exposure (task/stream metadata only)

Polling Required

The API does not support WebSockets or Server-Sent Events. Clients must poll for state changes.

Best Practices:

• Use reasonable poll intervals (5-10 seconds)
• Implement exponential backoff for errors
• Cache responses to reduce load

Performance Considerations

Response Sizes

Concurrency

The Fastify server handles concurrent requests efficiently. No special client-side throttling needed for typical orchestration workloads.

Database Impact

All endpoints query the SQLite database. Impact is minimal for typical usage patterns:

• Simple indexed queries (by stream ID, status, etc.)
• No complex joins or aggregations
• Database is shared with MCP server (read-only queries don't block)

Troubleshooting

Server Not Starting

Symptom: Logs show "Failed to start HTTP API server"

Causes:

1. Port already in use
2. Insufficient permissions
3. Invalid host address

Solution:

# Check if port is in use
lsof -i :9090

# Try a different port
# Edit .mcp.json:
"HTTP_API_PORT": "9091"

Empty Response

Symptom: Endpoints return empty arrays or no data

Causes:

1. No tasks/streams in current initiative
2. Archived streams (need includeArchived=true)
3. Filters too restrictive

Solution:

# Check initiative context
response = requests.get('http://localhost:9090/api/streams?includeArchived=true')

# Try broader filters
response = requests.get('http://localhost:9090/api/tasks')  # No filters

Stale Data

Symptom: Activity shows agents as idle when they're working

Cause: Agent heartbeat not updated (may indicate agent issue)

Solution: Check agent logs and task status. Activity heartbeat updates are managed by agents via the tc CLI.

See Also

• Task Management: tc CLI tool
• Stream Management: Stream orchestration guide (docs/orchestration/stream-management.md)