mcp-wcgw

Instructions

Implementation Reference

• src/wcgw/client/bash_state/bash_state.py:1184-1458 (handler)

  Core handler that executes BashCommand by interacting with pexpect.spawn shell process, handles sending commands/text/special keys, status checks, incremental output rendering, timeouts, background commands, etc.

  def execute_bash(
      bash_state: BashState,
      enc: EncoderDecoder[int],
      bash_arg: BashCommand,
      max_tokens: Optional[int],  # This will be noncoding_max_tokens
      timeout_s: Optional[float],
  ) -> tuple[str, float]:
      try:
          # Check if the thread_id matches current
          if bash_arg.thread_id != bash_state.current_thread_id:
              # Try to load state from the thread_id
              if not bash_state.load_state_from_thread_id(bash_arg.thread_id):
                  return (
                      f"Error: No saved bash state found for thread_id `{bash_arg.thread_id}`. Please initialize first with this ID.",
                      0.0,
                  )
  
          output, cost = _execute_bash(bash_state, enc, bash_arg, max_tokens, timeout_s)
  
          # Remove echo if it's a command
          if isinstance(bash_arg.action_json, Command):
              command = bash_arg.action_json.command.strip()
              if output.startswith(command):
                  output = output[len(command) :]
  
      finally:
          bash_state.run_bg_expect_thread()
          if bash_state.over_screen:
              thread = threading.Thread(
                  target=cleanup_orphaned_wcgw_screens,
                  args=(bash_state.console,),
                  daemon=True,
              )
              thread.start()
      return output, cost
  
  
  def assert_single_statement(command: str) -> None:
      # Check for multiple statements using the bash statement parser
      if "\n" in command:
          try:
              parser = BashStatementParser()
              statements = parser.parse_string(command)
          except Exception:
              # Fall back to simple newline check if something goes wrong
              raise ValueError(
                  "Command should not contain newline character in middle. Run only one command at a time."
              )
          if len(statements) > 1:
              raise ValueError(
                  "Error: Command contains multiple statements. Please run only one bash statement at a time."
              )
  
  
  def get_bg_running_commandsinfo(bash_state: BashState) -> str:
      msg = ""
      running = []
      for id_, state in bash_state.background_shells.items():
          running.append(f"Command: {state.last_command}, bg_command_id: {id_}")
      if running:
          msg = (
              "Following background commands are attached:\n" + "\n".join(running) + "\n"
          )
      else:
          msg = "No command running in background.\n"
      return msg
  
  
  def _execute_bash(
      bash_state: BashState,
      enc: EncoderDecoder[int],
      bash_arg: BashCommand,
      max_tokens: Optional[int],  # This will be noncoding_max_tokens
      timeout_s: Optional[float],
  ) -> tuple[str, float]:
      try:
          is_interrupt = False
          command_data = bash_arg.action_json
          is_bg = False
          og_bash_state = bash_state
  
          if not isinstance(command_data, Command) and command_data.bg_command_id:
              if command_data.bg_command_id not in bash_state.background_shells:
                  error = f"No shell found running with command id {command_data.bg_command_id}.\n"
                  if bash_state.background_shells:
                      error += get_bg_running_commandsinfo(bash_state)
                  if bash_state.state == "pending":
                      error += f"On the main thread a command is already running ({bash_state.last_command})"
                  else:
                      error += "On the main thread no command is running."
                  raise Exception(error)
              bash_state = bash_state.background_shells[command_data.bg_command_id]
              is_bg = True
  
          if isinstance(command_data, Command):
              if bash_state.bash_command_mode.allowed_commands == "none":
                  return "Error: BashCommand not allowed in current mode", 0.0
  
              bash_state.console.print(f"$ {command_data.command}")
  
              command = command_data.command.strip()
  
              assert_single_statement(command)
  
              if command_data.is_background:
                  bash_state = bash_state.start_new_bg_shell(bash_state.cwd)
                  is_bg = True
  
              if bash_state.state == "pending":
                  raise ValueError(WAITING_INPUT_MESSAGE)
  
              bash_state.clear_to_run()
              for i in range(0, len(command), 64):
                  bash_state.send(command[i : i + 64], set_as_command=None)
              bash_state.send(bash_state.linesep, set_as_command=command)
          elif isinstance(command_data, StatusCheck):
              bash_state.console.print("Checking status")
              if bash_state.state != "pending":
                  error = "No running command to check status of.\n"
                  error += get_bg_running_commandsinfo(bash_state)
                  return error, 0.0
  
          elif isinstance(command_data, SendText):
              if not command_data.send_text:
                  return "Failure: send_text cannot be empty", 0.0
  
              bash_state.console.print(f"Interact text: {command_data.send_text}")
              for i in range(0, len(command_data.send_text), 128):
                  bash_state.send(
                      command_data.send_text[i : i + 128], set_as_command=None
                  )
              bash_state.send(bash_state.linesep, set_as_command=None)
  
          elif isinstance(command_data, SendSpecials):
              if not command_data.send_specials:
                  return "Failure: send_specials cannot be empty", 0.0
  
              bash_state.console.print(
                  f"Sending special sequence: {command_data.send_specials}"
              )
              for char in command_data.send_specials:
                  if char == "Key-up":
                      bash_state.send("\033[A", set_as_command=None)
                  elif char == "Key-down":
                      bash_state.send("\033[B", set_as_command=None)
                  elif char == "Key-left":
                      bash_state.send("\033[D", set_as_command=None)
                  elif char == "Key-right":
                      bash_state.send("\033[C", set_as_command=None)
                  elif char == "Enter":
                      bash_state.send("\x0d", set_as_command=None)
                  elif char == "Ctrl-c":
                      bash_state.sendintr()
                      is_interrupt = True
                  elif char == "Ctrl-d":
                      bash_state.sendintr()
                      is_interrupt = True
                  elif char == "Ctrl-z":
                      bash_state.send("\x1a", set_as_command=None)
                  else:
                      raise Exception(f"Unknown special character: {char}")
  
          elif isinstance(command_data, SendAscii):
              if not command_data.send_ascii:
                  return "Failure: send_ascii cannot be empty", 0.0
  
              bash_state.console.print(
                  f"Sending ASCII sequence: {command_data.send_ascii}"
              )
              for ascii_char in command_data.send_ascii:
                  bash_state.send(chr(ascii_char), set_as_command=None)
                  if ascii_char == 3:
                      is_interrupt = True
          else:
              raise ValueError(f"Unknown command type: {type(command_data)}")
  
      except KeyboardInterrupt:
          bash_state.sendintr()
          bash_state.expect(bash_state.prompt)
          return "---\n\nFailure: user interrupted the execution", 0.0
  
      wait = min(timeout_s or CONFIG.timeout, CONFIG.timeout_while_output)
      index = bash_state.expect([bash_state.prompt, pexpect.TIMEOUT], timeout=wait)
      if index == 1:
          text = bash_state.before or ""
          incremental_text = _incremental_text(text, bash_state.pending_output)
  
          second_wait_success = False
          if is_status_check(bash_arg):
              # There's some text in BashInteraction mode wait for TIMEOUT_WHILE_OUTPUT
              remaining = CONFIG.timeout_while_output - wait
              patience = CONFIG.output_wait_patience
              if not incremental_text:
                  patience -= 1
              itext = incremental_text
              while remaining > 0 and patience > 0:
                  index = bash_state.expect(
                      [bash_state.prompt, pexpect.TIMEOUT], timeout=wait
                  )
                  if index == 0:
                      second_wait_success = True
                      break
                  else:
                      _itext = bash_state.before or ""
                      _itext = _incremental_text(_itext, bash_state.pending_output)
                      if _itext != itext:
                          patience = 3
                      else:
                          patience -= 1
                      itext = _itext
  
                  remaining = remaining - wait
  
              if not second_wait_success:
                  text = bash_state.before or ""
                  incremental_text = _incremental_text(text, bash_state.pending_output)
  
          if not second_wait_success:
              bash_state.set_pending(text)
  
              tokens = enc.encoder(incremental_text)
  
              if max_tokens and len(tokens) >= max_tokens:
                  incremental_text = "(...truncated)\n" + enc.decoder(
                      tokens[-(max_tokens - 1) :]
                  )
  
              if is_interrupt:
                  incremental_text = (
                      incremental_text
                      + """---
  ----
  Failure interrupting.
  You may want to try Ctrl-c again or program specific exit interactive commands.
      """
                  )
  
              exit_status = get_status(bash_state, is_bg)
              incremental_text += exit_status
              if is_bg and bash_state.state == "repl":
                  try:
                      bash_state.cleanup()
                      og_bash_state.background_shells.pop(bash_state.current_thread_id)
                  except Exception as e:
                      bash_state.console.log(f"error while cleaning up {e}")
  
              return incremental_text, 0
  
      before = str(bash_state.before)
  
      output = _incremental_text(before, bash_state.pending_output)
      bash_state.set_repl()
  
      tokens = enc.encoder(output)
      if max_tokens and len(tokens) >= max_tokens:
          output = "(...truncated)\n" + enc.decoder(tokens[-(max_tokens - 1) :])
  
      try:
          exit_status = get_status(bash_state, is_bg)
          output += exit_status
          if is_bg and bash_state.state == "repl":
              try:
                  bash_state.cleanup()
                  og_bash_state.background_shells.pop(bash_state.current_thread_id)
              except Exception as e:
                  bash_state.console.log(f"error while cleaning up {e}")
      except ValueError:
          bash_state.console.print(output)
          bash_state.console.print(traceback.format_exc())
          bash_state.console.print("Malformed output, restarting shell", style="red")
          # Malformed output, restart shell
          bash_state.reset_shell()
          output = "(exit shell has restarted)"
      return output, 0

• src/wcgw/client/mcp_server/server.py:94-148 (handler)

  MCP server tool handler that parses arguments into BashCommand model and calls get_tool_output for execution.

  @server.call_tool()  # type: ignore
  async def handle_call_tool(
      name: str, arguments: dict[str, Any] | None
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      global BASH_STATE
      if not arguments:
          raise ValueError("Missing arguments")
  
      tool_type = which_tool_name(name)
      tool_call = parse_tool_by_name(name, arguments)
  
      try:
          assert BASH_STATE
          output_or_dones, _ = get_tool_output(
              Context(BASH_STATE, BASH_STATE.console),
              tool_call,
              default_enc,
              0.0,
              lambda x, y: ("", 0),
              24000,  # coding_max_tokens
              8000,  # noncoding_max_tokens
          )
  
      except Exception as e:
          output_or_dones = [f"GOT EXCEPTION while calling tool. Error: {e}"]
  
      content: list[types.TextContent | types.ImageContent | types.EmbeddedResource] = []
      for output_or_done in output_or_dones:
          if isinstance(output_or_done, str):
              if issubclass(tool_type, Initialize):
                  # Prepare the original hardcoded message
                  original_message = """
  - Additional important note: as soon as you encounter "The user has chosen to disallow the tool call.", immediately stop doing everything and ask user for the reason.
  
  Initialize call done.
      """
  
                  # If custom instructions exist, prepend them to the original message
                  if CUSTOM_INSTRUCTIONS:
                      output_or_done += f"\n{CUSTOM_INSTRUCTIONS}\n{original_message}"
                  else:
                      output_or_done += original_message
  
              content.append(types.TextContent(type="text", text=output_or_done))
          else:
              content.append(
                  types.ImageContent(
                      type="image",
                      data=output_or_done.data,
                      mimeType=output_or_done.media_type,
                  )
              )
  
      return content

• src/wcgw/types_.py:159-177 (schema)

  Pydantic models defining the input schema for BashCommand tool, using ActionJsonSchema union types, overrides JSON schema to use BashCommandOverride for LLM compatibility.

  class BashCommandOverride(BaseModel):
      action_json: ActionJsonSchema
      wait_for_seconds: Optional[float] = None
      thread_id: str
  
  
  class BashCommand(BaseModel):
      action_json: Command | StatusCheck | SendText | SendSpecials | SendAscii
      wait_for_seconds: Optional[float] = None
      thread_id: str
  
      def model_post_init(self, __context: Any) -> None:
          self.thread_id = normalize_thread_id(self.thread_id)
          return super().model_post_init(__context)
  
      @staticmethod
      def model_json_schema(*args, **kwargs) -> dict[str, Any]:  # type: ignore
          return BashCommandOverride.model_json_schema(*args, **kwargs)

• src/wcgw/client/mcp_server/server.py:84-92 (registration)

  MCP server registration: lists all tools including BashCommand using pre-defined TOOL_PROMPTS.

  @server.list_tools()  # type: ignore
  async def handle_list_tools() -> list[types.Tool]:
      """
      List available tools.
      Each tool specifies its arguments using JSON Schema validation.
      """
  
      return TOOL_PROMPTS

• src/wcgw/client/tool_prompts.py:37-55 (registration)

  Registers BashCommand tool schema, description, and annotations in TOOL_PROMPTS used by MCP server.

      Tool(
          inputSchema=remove_titles_from_schema(BashCommand.model_json_schema()),
          name="BashCommand",
          description="""
  - Execute a bash command. This is stateful (beware with subsequent calls).
  - Status of the command and the current working directory will always be returned at the end.
  - The first or the last line might be `(...truncated)` if the output is too long.
  - Always run `pwd` if you get any file or directory not found error to make sure you're not lost.
  - Do not run bg commands using "&", instead use this tool.
  - You must not use echo/cat to read/write files, use ReadFiles/FileWriteOrEdit
  - In order to check status of previous command, use `status_check` with empty command argument.
  - Only command is allowed to run at a time. You need to wait for any previous command to finish before running a new one.
  - Programs don't hang easily, so most likely explanation for no output is usually that the program is still running, and you need to check status again.
  - Do not send Ctrl-c before checking for status till 10 minutes or whatever is appropriate for the program to finish.
  - Only run long running commands in background. Each background command is run in a new non-reusable shell.
  - On running a bg command you'll get a bg command id that you should use to get status or interact.
  """,
          annotations=ToolAnnotations(destructiveHint=True, openWorldHint=True),
      ),