kiro_task_status
Check status and retrieve partial results from async tasks for polling streaming responses in Kiro CLI workflows.
Instructions
Get status and partial results of an async task. Use for polling streaming results.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_id | Yes | The task ID returned by kiro_chat_async | |
| from_chunk_index | No | Get chunks starting from this index (for incremental updates) |
Implementation Reference
- src/kiro_cli_mcp/server.py:353-370 (handler)The main handler function for the kiro_task_status tool, which extracts arguments and calls the task manager to get status.
async def _handle_task_status( task_manager: StreamingTaskManager, arguments: dict[str, Any] ) -> dict[str, Any]: """Handle kiro_task_status tool call - get task status and partial results.""" task_id = arguments.get("task_id", "") from_chunk_index = arguments.get("from_chunk_index", 0) status = await task_manager.get_task_status(task_id, from_chunk_index) if status is None: return { "error": "Task not found", "task_id": task_id, } return status - src/kiro_cli_mcp/tools.py:158-176 (schema)The input schema definition for the kiro_task_status tool, defining required task_id and optional from_chunk_index.
{ "name": "kiro_task_status", "description": "Get status and partial results of an async task. Use for polling streaming results.", "inputSchema": { "type": "object", "properties": { "task_id": { "type": "string", "description": "The task ID returned by kiro_chat_async" }, "from_chunk_index": { "type": "integer", "description": "Get chunks starting from this index (for incremental updates)", "default": 0 } }, "required": ["task_id"] } }, - src/kiro_cli_mcp/server.py:119-120 (registration)Dispatch/registration point in the call_tool handler that routes kiro_task_status calls to the specific handler.
elif name == "kiro_task_status": result = await _handle_task_status(task_manager, arguments) - Supporting method in StreamingTaskManager that implements the status retrieval logic used by the tool handler.
async def get_task_status( self, task_id: str, from_chunk_index: int = 0, ) -> dict[str, Any] | None: """Get task status with new chunks since last poll. Args: task_id: Task ID to check from_chunk_index: Get chunks starting from this index Returns: Status dict or None if task not found """ task = self._tasks.get(task_id) if task is None: return None return task.to_status_dict(from_chunk_index)