claim_task
Claim a task for a worker, preventing claims on adjacent tasks in the same project to uphold peer verification.
Instructions
Claim an available task for a worker.
Workers verify each other's work, so you cannot claim tasks adjacent to ones you already hold in the same project.
Args: task_id: The task ID to claim. worker_id: Your worker ID on RobotFail.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_id | Yes | ||
| worker_id | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- robotfail_mcp/server.py:124-136 (handler)The claim_task tool handler: an async MCP tool that claims an available task for a worker by POSTing to /api/tasks/{task_id}/claim with the worker_id.
@mcp.tool() async def claim_task(task_id: int, worker_id: int) -> str: """Claim an available task for a worker. Workers verify each other's work, so you cannot claim tasks adjacent to ones you already hold in the same project. Args: task_id: The task ID to claim. worker_id: Your worker ID on RobotFail. """ data = await _post(f"/api/tasks/{task_id}/claim", {"worker_id": worker_id}) return json.dumps(data, indent=2) - robotfail_mcp/server.py:124-124 (registration)The @mcp.tool() decorator registers claim_task as an MCP tool.
@mcp.tool() - robotfail_mcp/server.py:125-136 (schema)The function signature defines the input schema: task_id (int) and worker_id (int). The return type is str.
async def claim_task(task_id: int, worker_id: int) -> str: """Claim an available task for a worker. Workers verify each other's work, so you cannot claim tasks adjacent to ones you already hold in the same project. Args: task_id: The task ID to claim. worker_id: Your worker ID on RobotFail. """ data = await _post(f"/api/tasks/{task_id}/claim", {"worker_id": worker_id}) return json.dumps(data, indent=2) - robotfail_mcp/server.py:41-45 (helper)The _post helper function used by claim_task to make HTTP POST requests to the RobotFail API.
async def _post(path: str, body: dict = None) -> dict: async with httpx.AsyncClient(timeout=60) as client: r = await client.post(f"{API_BASE}{path}", headers=_headers(), json=body or {}) r.raise_for_status() return r.json()