Instructions

Implementation Reference

• mcp_claude_code/tools/shell/run_command.py:274-328 (handler)

  Core handler logic for the run_command tool: extracts parameters, checks permissions, executes via BashSessionExecutor, and formats results.
  @override async def call( self, ctx: MCPContext, **params: Unpack[RunCommandToolParams], ) -> str: """Execute the tool with the given parameters. Args: ctx: MCP context **params: Tool parameters Returns: Tool result """ tool_ctx = await self.prepare_tool_context(ctx) # Extract parameters command = params["command"] session_id = params.get("session_id") time_out = params.get("time_out", 30) is_input = params.get("is_input", False) blocking = params.get("blocking", False) await tool_ctx.info( f"Executing command: {command} ( session_id={session_id}, is_input={is_input}, blocking={blocking} )" ) # Check if command is allowed (skip for input to running processes) if not is_input and not self.command_executor.is_command_allowed(command): await tool_ctx.error(f"Command not allowed: {command}") return f"Error: Command not allowed: {command}" try: # Execute command using BashSessionExecutor with enhanced parameters result = await self.command_executor.execute_command( command=command, timeout=time_out, session_id=session_id, is_input=is_input, blocking=blocking, ) except RuntimeError as e: await tool_ctx.error(f"Session execution failed: {str(e)}") return f"Error: Session execution failed: {str(e)}" # Format the result using the new enhanced formatting if result.is_success: # Use the enhanced agent observation format for better user experience return result.to_agent_observation() else: # For failed or running commands, provide comprehensive output return result.format_output() @override
• mcp_claude_code/tools/shell/run_command.py:59-75 (schema)

  TypedDict defining the input parameters and annotations for the run_command tool.
  class RunCommandToolParams(TypedDict): """Parameters for the RunCommandTool. Attributes: command: The shell command to execute session_id: Optional session ID for persistent shell sessions time_out: Timeout in seconds for session-based commands with no output changes is_input: Whether this is input to a running process rather than a new command blocking: Whether to run in blocking mode, ignoring no-change timeout """ command: Command session_id: SessionID time_out: TimeOut is_input: IsInput blocking: Blocking
• mcp_claude_code/tools/shell/run_command.py:329-358 (registration)

  The register method defines and registers the run_command tool function using @mcp_server.tool decorator, which serves as the MCP entrypoint.
  def register(self, mcp_server: FastMCP) -> None: """Register this run command tool with the MCP server. Creates a wrapper function with explicitly defined parameters that match the tool's parameter schema and registers it with the MCP server. Args: mcp_server: The FastMCP server instance """ tool_self = self # Create a reference to self for use in the closure @mcp_server.tool(name=self.name, description=self.description) @handle_connection_errors async def run_command( command: Command, session_id: SessionID, time_out: TimeOut, is_input: IsInput, blocking: Blocking, ) -> str: ctx = get_context() return await tool_self.call( ctx, command=command, session_id=session_id, time_out=time_out, is_input=is_input, blocking=blocking, )
• mcp_claude_code/tools/shell/__init__.py:52-68 (registration)

  Registers all shell tools, including run_command, by creating instances via get_shell_tools and calling ToolRegistry.register_tools (which invokes each tool's register method).
  def register_shell_tools( mcp_server: FastMCP, permission_manager: PermissionManager, ) -> list[BaseTool]: """Register all shell tools with the MCP server. Args: mcp_server: The FastMCP server instance permission_manager: Permission manager for access control Returns: List of registered tools """ tools = get_shell_tools(permission_manager) ToolRegistry.register_tools(mcp_server, tools) return tools
• mcp_claude_code/tools/shell/bash_session_executor.py:123-221 (helper)

  Supporting executor that performs the actual command execution using BashSession or subprocess, called by the run_command handler.
  async def execute_command( self, command: str, env: dict[str, str] | None = None, timeout: float | None = 60.0, session_id: str = "", is_input: bool = False, blocking: bool = False, ) -> CommandResult: """Execute a shell command with safety checks. Args: command: The command to execute env: Optional environment variables timeout: Optional timeout in seconds session_id: Optional session ID for persistent execution is_input: Whether this is input to a running process blocking: Whether to run in blocking mode Returns: CommandResult containing execution results """ self._log( f"Executing command: {command} (is_input={is_input}, blocking={blocking})" ) # Check if the command is allowed (skip for input to running processes) if not is_input and not self.is_command_allowed(command): return CommandResult( return_code=1, error_message=f"Command not allowed: {command}", session_id=session_id, command=command, ) # Handle subprocess mode when session_id is explicitly None if not session_id: return await self._execute_subprocess_mode(command, env, timeout) # Default working directory for new sessions only # Existing sessions maintain their current working directory # Users should use 'cd' commands to navigate within sessions default_work_dir = os.path.expanduser("~") # Generate default session ID if none provided effective_session_id = session_id if effective_session_id == "": import uuid effective_session_id = f"default_{uuid.uuid4().hex[:8]}" self._log(f"Generated default session ID: {effective_session_id}") # Use session-based execution try: # Get existing session or create new one session = self.session_manager.get_session(effective_session_id) if session is None: # Use faster timeouts and polling for tests if self.fast_test_mode: timeout_seconds = ( 10 # Faster timeout for tests but not too aggressive ) poll_interval = 0.2 # Faster polling for tests but still reasonable else: timeout_seconds = 30 # Default timeout poll_interval = 0.5 # Default polling session = self.session_manager.get_or_create_session( session_id=effective_session_id, work_dir=default_work_dir, no_change_timeout_seconds=timeout_seconds, poll_interval=poll_interval, ) # Set environment variables if provided (only for new commands, not input) if env and not is_input: for key, value in env.items(): env_result = session.execute(f'export {key}="{value}"') if env_result.return_code != 0: self._log(f"Failed to set environment variable {key}") # Execute the command with enhanced parameters result = session.execute( command=command, is_input=is_input, blocking=blocking, timeout=timeout ) # Add session_id to the result result.session_id = effective_session_id return result except Exception as e: self._log(f"Session execution error: {str(e)}") return CommandResult( return_code=1, error_message=f"Error executing command in session: {str(e)}", session_id=effective_session_id, command=command, ) async def _execute_subprocess_mode(