terminate_process
Terminate an interactive process by session ID with optional force kill (SIGKILL) and configurable grace period before escalation.
Instructions
Terminate an interactive process.
Args: session_id: The session ID to terminate. force: Use SIGKILL instead of SIGTERM. Default False. grace_period: Seconds to wait after SIGTERM before SIGKILL. Default 5.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes | ||
| force | No | ||
| grace_period | No |
Implementation Reference
- Top-level MCP tool handler for terminate_process, registered via @mcp.tool() decorator. Fetches the session and delegates to session.terminate().
@mcp.tool() def terminate_process( session_id: str, force: bool = False, grace_period: float = 5.0, ) -> dict: """Terminate an interactive process. Args: session_id: The session ID to terminate. force: Use SIGKILL instead of SIGTERM. Default False. grace_period: Seconds to wait after SIGTERM before SIGKILL. Default 5. """ session = _mgr.get(session_id) if not session: return {"error": f"Session '{session_id}' not found"} session.terminate(force=force, grace_period=grace_period) return {"success": True} - Alternate handler in the tools-based registration. Delegates to SessionManager.terminate().
def terminate_process(args: dict) -> dict: mgr.terminate( args["session_id"], force=args.get("force", False), grace_period=args.get("grace_period", 5.0), ) return {"success": True} - src/interactive_process_mcp/tools.py:98-100 (registration)Registration of terminate_process in the tools tuple list returned by create_tools().
("terminate_process", terminate_process), ("resize_pty", resize_pty), ("get_session_info", get_session_info), - SessionManager.terminate() — thread-safe lookup and delegation to Session.terminate().
def terminate(self, session_id: str, force: bool = False, grace_period: float = 5.0) -> None: with self._lock: session = self._sessions.get(session_id) if session: session.terminate(force=force, grace_period=grace_period) - Session.terminate() — core termination logic: handles PTY mode (pexpect spawn) and pipe mode (subprocess.Popen), with force/grace_period semantics.
def terminate(self, force: bool = False, grace_period: float = 5.0) -> None: if self.status != SessionStatus.RUNNING: return if self.mode == "pty": if force: self._process.terminate(force=True) else: self._process.terminate(force=False) import time time.sleep(grace_period) if self._process.isalive(): self._process.terminate(force=True) else: if force: self._process.kill() else: self._process.send_signal(signal.SIGTERM) try: self._process.wait(timeout=grace_period) except subprocess.TimeoutExpired: self._process.kill() if self._reader_thread: self._reader_thread.join(timeout=2.0) self.status = SessionStatus.EXITED