Agent Skills: Task-Router MCP Server

Task-Router MCP Server

Purpose

Coordinates multi-agent workflows via a task queue, eliminating manual handoffs and ensuring autonomous execution.

When to Use

• When delegating long-running or complex tasks to specialized agents (e.g., Gemini, Codex).
• When orchestrating multi-step workflows where output from one agent is input for another.

Process

1. Task Creation: Orchestrator (Claude Desktop) creates tasks with specific inputs.
2. Polling: Agents poll the JSON-based queue for pending tasks.
3. Claiming: An agent marks a task as in_progress.
4. Execution: The agent performs the task and outputs results.
5. Completion: The agent reports completion, triggering the next task in the chain.

Architecture

Claude Desktop (Orchestrator)
    ↓ creates tasks
Task Queue (JSON file-based)
    ↓ agents poll
[Gemini | Claude Code | Codex CLI]
    ↓ claim → execute → complete
Next Task Auto-assigned

MCP Tools

1. create_task

{
  "task_id": "asset-3-generation",
  "assigned_to": "gemini",
  "status": "pending",
  "priority": "high",
  "inputs": {
    "prompt": "[full prompt text]",
    "resolution": "512x512"
  },
  "outputs": null,
  "next_task": "asset-3-validation",
  "next_assigned_to": "claude-desktop"
}

2. claim_task

Agent marks task as in-progress:

{
  "task_id": "asset-3-generation",
  "status": "in_progress",
  "claimed_by": "gemini",
  "claimed_at": "2026-01-30T10:15:00Z"
}

3. complete_task

Agent reports completion:

{
  "task_id": "asset-3-generation",
  "status": "completed",
  "outputs": {
    "image_path": "/downloads/asset-3-attempt-2.png",
    "notes": "Generated successfully"
  },
  "completed_at": "2026-01-30T10:18:00Z"
}

4. list_tasks

Query tasks by status/agent:

{
  "status": "pending",
  "assigned_to": "gemini"
}

Returns array of matching tasks.

5. rollback_task

If validation fails, rollback to previous state:

{
  "task_id": "asset-3-validation",
  "rollback_to": "asset-3-generation",
  "reason": "Score 85 (below 90 threshold)"
}

Task Flow Example

Phase: Asset 3 Generation

1. Claude Desktop creates task:

{
  "task_id": "asset-3-gen",
  "assigned_to": "gemini",
  "inputs": { "prompt": "...", "resolution": "512x512" },
  "next_task": "asset-3-validate"
}

2. Gemini polls queue → claims task → generates → completes:

{
  "status": "completed",
  "outputs": { "image_path": "/downloads/asset-3.png" }
}

3. Router auto-creates next task:

{
  "task_id": "asset-3-validate",
  "assigned_to": "claude-desktop",
  "inputs": { "image_path": "/downloads/asset-3.png" }
}

4. Claude Desktop claims → validates → scores 92 → completes
5. Router auto-creates packaging task for Claude Code

Implementation

File: /servers/task_router_mcp.py

Queue Storage: /tmp/kerala-rage-task-queue.json

class TaskRouter:
    def __init__(self):
        self.queue_file = "/tmp/kerala-rage-task-queue.json"
        self.tasks = self.load_queue()

    def create_task(self, task_data):
        task = {
            "task_id": task_data['task_id'],
            "assigned_to": task_data['assigned_to'],
            "status": "pending",
            "created_at": datetime.now().isoformat(),
            **task_data
        }
        self.tasks.append(task)
        self.save_queue()
        return task

    def claim_task(self, task_id, agent):
        task = self.get_task(task_id)
        task['status'] = 'in_progress'
        task['claimed_by'] = agent
        task['claimed_at'] = datetime.now().isoformat()
        self.save_queue()

    def complete_task(self, task_id, outputs):
        task = self.get_task(task_id)
        task['status'] = 'completed'
        task['outputs'] = outputs
        task['completed_at'] = datetime.now().isoformat()

        # Auto-create next task if specified
        if task.get('next_task'):
            self.create_task({
                'task_id': task['next_task'],
                'assigned_to': task['next_assigned_to'],
                'inputs': outputs  # Previous outputs → next inputs
            })

        self.save_queue()

Integration

Claude Desktop Config:

{
  "mcpServers": {
    "task-router": {
      "command": "python3",
      "args": ["/path/to/task_router_mcp.py"]
    }
  }
}

Agent Polling

Each agent polls queue every 30 seconds:

Gemini (Antigravity):

while True:
    tasks = mcp.call_tool("task-router", "list_tasks", {
        "status": "pending",
        "assigned_to": "gemini"
    })
    if tasks:
        task = tasks[0]
        mcp.call_tool("task-router", "claim_task", {"task_id": task['task_id']})
        # Execute generation
        mcp.call_tool("task-router", "complete_task", {
            "task_id": task['task_id'],
            "outputs": {"image_path": result_path}
        })
    time.sleep(30)

Progress Tracking

Dashboard view:

{
  "total_tasks": 47,
  "completed": 23,
  "in_progress": 3,
  "pending": 21,
  "by_agent": {
    "gemini": { "completed": 10, "in_progress": 1 },
    "claude-code": { "completed": 8, "in_progress": 2 },
    "codex": { "completed": 5, "in_progress": 0 }
  }
}

Benefits

• Eliminates manual handoffs
• Progress tracking built-in
• Rollback capability
• Agent autonomy (claim tasks independently)
• Context preserved in task chain

Queue-based orchestration. Manual coordination → automatic workflow.