Kiro CLI MCP Server

Instructions

Implementation Reference

• src/kiro_cli_mcp/server.py:323-351 (handler)

  The primary handler function for the kiro_chat_async tool. It creates a session if needed, sets up a task executor, starts a background task via StreamingTaskManager, and returns the task ID for polling.

  async def _handle_chat_async(
      session_manager: SessionManager,
      command_executor: CommandExecutor,
      task_manager: StreamingTaskManager,
      arguments: dict[str, Any]
  ) -> dict[str, Any]:
      """Handle kiro_chat_async tool call - start async chat with polling support."""
      message = arguments.get("message", "")
      session_id = arguments.get("session_id")
      
      session = await session_manager.get_or_create_session(session_id)
      
      # Create task executor
      executor = create_chat_task_executor(command_executor, session)
      
      # Start async task
      task = await task_manager.start_task(
          session_id=session.id,
          message=message,
          executor=executor,
      )
      
      return {
          "task_id": task.task_id,
          "session_id": session.id,
          "status": task.status.value,
          "message": "Task started. Poll kiro_task_status for updates.",
      }

• src/kiro_cli_mcp/tools.py:140-157 (schema)

  JSON Schema definition for the kiro_chat_async tool, including input parameters and description used for MCP tool listing and validation.

  {
      "name": "kiro_chat_async",
      "description": "Start an async chat task for streaming-like experience. Use kiro_task_status to poll for results.",
      "inputSchema": {
          "type": "object",
          "properties": {
              "message": {
                  "type": "string",
                  "description": "The message to send to kiro-cli"
              },
              "session_id": {
                  "type": "string",
                  "description": "Optional session ID. Uses active session if not provided"
              }
          },
          "required": ["message"]
      }
  },

• src/kiro_cli_mcp/server.py:115-119 (registration)

  Dispatch routing in the main call_tool handler that directs kiro_chat_async calls to the specific _handle_chat_async function.

  elif name == "kiro_chat_async":
      result = await _handle_chat_async(
          session_manager, command_executor, task_manager, arguments
      )
  elif name == "kiro_task_status":

• src/kiro_cli_mcp/streaming.py:466-484 (helper)

  Helper function called by the handler to create the TaskExecutor (StreamingChatExecutor wrapper) that performs the actual async chat execution with progress chunking.

  def create_chat_task_executor(
      command_executor: Any,
      session: Any,
  ) -> TaskExecutor:
      """Create a task executor for chat.
      
      Args:
          command_executor: The CommandExecutor to use
          session: The Session to use
          
      Returns:
          A TaskExecutor function
      """
      streaming_executor = StreamingChatExecutor(command_executor)
      
      async def executor(task: AsyncTask) -> str:
          return await streaming_executor.execute(task, session)
      
      return executor