Kiro CLI MCP Server

Instructions

Implementation Reference

• src/kiro_cli_mcp/server.py:252-263 (handler)

  Main handler function for kiro_session_end tool that extracts the session_id from arguments and delegates to SessionManager.end_session(session_id)

  async def _handle_session_end(
      session_manager: SessionManager,
      arguments: dict[str, Any]
  ) -> dict[str, Any]:
      """Handle kiro_session_end tool call."""
      session_id = arguments.get("session_id", "")
      await session_manager.end_session(session_id)
      return {
          "success": True,
          "session_id": session_id,
      }

• src/kiro_cli_mcp/tools.py:69-82 (schema)

  Input schema definition for the kiro_session_end tool, specifying required session_id parameter

  {
      "name": "kiro_session_end",
      "description": "End a kiro-cli session",
      "inputSchema": {
          "type": "object",
          "properties": {
              "session_id": {
                  "type": "string",
                  "description": "The session ID to end"
              }
          },
          "required": ["session_id"]
      }
  },

• src/kiro_cli_mcp/server.py:104-105 (registration)

  Registration/dispatch of kiro_session_end tool call within the main handle_call_tool function

  elif name == "kiro_session_end":
      result = await _handle_session_end(session_manager, arguments)

• src/kiro_cli_mcp/session.py:410-454 (helper)

  SessionManager.end_session method implementing the core logic to terminate the kiro-cli process and remove the session

  async def end_session(self, session_id: str) -> bool:
      """End and cleanup a session.
      
      Args:
          session_id: The session ID to end
          
      Returns:
          True if session was ended
          
      Raises:
          SessionError: If session not found
      """
      async with self._lock:
          session = self.sessions.get(session_id)
          if session is None:
              raise SessionError(
                  code=ErrorCode.SESSION_NOT_FOUND,
                  details={"session_id": session_id},
              )
          
          # Terminate process if running
          if session.process is not None:
              try:
                  session.process.terminate()
                  await asyncio.wait_for(
                      session.process.wait(),
                      timeout=5.0,
                  )
              except asyncio.TimeoutError:
                  session.process.kill()
              except Exception:
                  pass
          
          # Remove from sessions
          del self.sessions[session_id]
          
          # Update active session if needed
          if self.active_session_id == session_id:
              if self.sessions:
                  self.active_session_id = next(iter(self.sessions.keys()))
              else:
                  self.active_session_id = None
          
          return True