kiro_session_end
Terminate an active Kiro CLI session to free resources and manage workflows by specifying the session ID.
Instructions
End a kiro-cli session
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes | The session ID to end |
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