mcp-wcgw

Instructions

Implementation Reference

• src/wcgw/client/tools.py:105-342 (handler)

  Main handler function for the Initialize tool. Initializes the bash state, loads or creates workspace context, handles mode configuration, reads initial files, loads task memory if resuming, generates comprehensive system prompt including environment info, repo structure, alignment guidelines.

  def initialize(
      type: Literal["user_asked_change_workspace", "first_call"],
      context: Context,
      any_workspace_path: str,
      read_files_: list[str],
      task_id_to_resume: str,
      coding_max_tokens: Optional[int],
      noncoding_max_tokens: Optional[int],
      mode: ModesConfig,
      thread_id: str,
  ) -> tuple[str, Context, dict[str, list[tuple[int, int]]]]:
      # Expand the workspace path
      any_workspace_path = expand_user(any_workspace_path)
      repo_context = ""
  
      memory = ""
      loaded_state = None
  
      # For workspace/mode changes, ensure we're using an existing state if possible
      if type != "first_call" and thread_id != context.bash_state.current_thread_id:
          # Try to load state from the thread_id
          if not context.bash_state.load_state_from_thread_id(thread_id):
              return (
                  f"Error: No saved bash state found for thread_id `{thread_id}`. Please re-initialize to get a new id or use correct id.",
                  context,
                  {},
              )
      del (
          thread_id
      )  # No use other than loading correct state before doing actual tool related stuff
  
      # Handle task resumption - this applies only to first_call
      if type == "first_call" and task_id_to_resume:
          try:
              project_root_path, task_mem, loaded_state = load_memory(
                  task_id_to_resume,
                  coding_max_tokens,
                  noncoding_max_tokens,
                  lambda x: default_enc.encoder(x),
                  lambda x: default_enc.decoder(x),
              )
              memory = "Following is the retrieved task:\n" + task_mem
              if os.path.exists(project_root_path):
                  any_workspace_path = project_root_path
  
          except Exception:
              memory = f'Error: Unable to load task with ID "{task_id_to_resume}" '
      elif task_id_to_resume:
          memory = (
              "Warning: task can only be resumed in a new conversation. No task loaded."
          )
  
      folder_to_start = None
      if type == "first_call" and not any_workspace_path:
          tmp_dir = get_tmpdir()
          any_workspace_path = os.path.join(
              tmp_dir, "claude-playground-" + uuid.uuid4().hex[:4]
          )
  
      if any_workspace_path:
          if os.path.exists(any_workspace_path):
              if os.path.isfile(any_workspace_path):
                  # Set any_workspace_path to the directory containing the file
                  # Add the file to read_files_ only if empty to avoid duplicates
                  if not read_files_:
                      read_files_ = [any_workspace_path]
                  any_workspace_path = os.path.dirname(any_workspace_path)
              # Let get_repo_context handle loading the workspace stats
              repo_context, folder_to_start = get_repo_context(any_workspace_path)
  
              repo_context = f"---\n# Workspace structure\n{repo_context}\n---\n"
  
              # update modes if they're relative
              if isinstance(mode, CodeWriterMode):
                  mode.update_relative_globs(any_workspace_path)
              else:
                  assert isinstance(mode, str)
          else:
              if os.path.abspath(any_workspace_path):
                  os.makedirs(any_workspace_path, exist_ok=True)
                  repo_context = f"\nInfo: Workspace path {any_workspace_path} did not exist. I've created it for you.\n"
                  folder_to_start = Path(any_workspace_path)
              else:
                  repo_context = (
                      f"\nInfo: Workspace path {any_workspace_path} does not exist."
                  )
      # Restore bash state if available
      if loaded_state is not None:
          try:
              parsed_state = BashState.parse_state(loaded_state)
              workspace_root = (
                  str(folder_to_start) if folder_to_start else parsed_state[5]
              )
              loaded_thread_id = parsed_state[6] if len(parsed_state) > 6 else None
  
              if not loaded_thread_id:
                  loaded_thread_id = context.bash_state.current_thread_id
  
              if mode == "wcgw":
                  context.bash_state.load_state(
                      parsed_state[0],
                      parsed_state[1],
                      parsed_state[2],
                      parsed_state[3],
                      {**parsed_state[4], **context.bash_state.whitelist_for_overwrite},
                      str(folder_to_start) if folder_to_start else workspace_root,
                      workspace_root,
                      loaded_thread_id,
                  )
              else:
                  state = modes_to_state(mode)
                  context.bash_state.load_state(
                      state[0],
                      state[1],
                      state[2],
                      state[3],
                      {**parsed_state[4], **context.bash_state.whitelist_for_overwrite},
                      str(folder_to_start) if folder_to_start else workspace_root,
                      workspace_root,
                      loaded_thread_id,
                  )
          except ValueError:
              context.console.print(traceback.format_exc())
              context.console.print("Error: couldn't load bash state")
              pass
          mode_prompt = get_mode_prompt(context)
      else:
          mode_changed = is_mode_change(mode, context.bash_state)
          state = modes_to_state(mode)
          new_thread_id = context.bash_state.current_thread_id
          if type == "first_call":
              # Recreate thread_id
              new_thread_id = generate_thread_id()
          # Use the provided workspace path as the workspace root
          context.bash_state.load_state(
              state[0],
              state[1],
              state[2],
              state[3],
              dict(context.bash_state.whitelist_for_overwrite),
              str(folder_to_start) if folder_to_start else "",
              str(folder_to_start) if folder_to_start else "",
              new_thread_id,
          )
          if type == "first_call" or mode_changed:
              mode_prompt = get_mode_prompt(context)
          else:
              mode_prompt = ""
  
      del mode
  
      initial_files_context = ""
      initial_paths_with_ranges: dict[str, list[tuple[int, int]]] = {}
      if read_files_:
          if folder_to_start:
              read_files_ = [
                  # Expand the path before checking if it's absolute
                  os.path.join(folder_to_start, f)
                  if not os.path.isabs(expand_user(f))
                  else expand_user(f)
                  for f in read_files_
              ]
          initial_files, initial_paths_with_ranges, _ = read_files(
              read_files_, coding_max_tokens, noncoding_max_tokens, context
          )
          initial_files_context = f"---\n# Requested files\nHere are the contents of the requested files:\n{initial_files}\n---\n"
  
      # Check for global CLAUDE.md and workspace CLAUDE.md
      alignment_context = ""
  
      # Check if ripgrep is available and add instruction if it is
      try:
          subprocess.run(["which", "rg"], timeout=1, capture_output=True, check=True)
          alignment_context += "---\n# Available commands\n\n- Use ripgrep `rg` command instead of `grep` because it's much much faster.\n\n---\n\n"
      except Exception:
          pass
  
      # Check for global alignment doc in ~/.wcgw: prefer CLAUDE.md, else AGENTS.md
      try:
          global_dir = os.path.join(expanduser("~"), ".wcgw")
          for fname in ("CLAUDE.md", "AGENTS.md"):
              global_alignment_file_path = os.path.join(global_dir, fname)
              if os.path.exists(global_alignment_file_path):
                  with open(global_alignment_file_path, "r") as f:
                      global_alignment_content = f.read()
                  alignment_context += f"---\n# Important guidelines from the user\n```\n{global_alignment_content}\n```\n---\n\n"
                  break
      except Exception as e:
          # Log any errors when reading the global file
          context.console.log(f"Error reading global alignment file: {e}")
  
      # Then check for workspace-specific alignment doc: prefer CLAUDE.md, else AGENTS.md
      if folder_to_start:
          try:
              base_dir = str(folder_to_start)
              selected_name = ""
              alignment_content = ""
              for fname in ("CLAUDE.md", "AGENTS.md"):
                  alignment_file_path = os.path.join(base_dir, fname)
                  if os.path.exists(alignment_file_path):
                      with open(alignment_file_path, "r") as f:
                          alignment_content = f.read()
                      selected_name = fname
                      break
              if alignment_content:
                  alignment_context += f"---\n# {selected_name} - user shared project guidelines to follow\n```\n{alignment_content}\n```\n---\n\n"
          except Exception as e:
              # Log any errors when reading the workspace file
              context.console.log(f"Error reading workspace alignment file: {e}")
  
      uname_sysname = os.uname().sysname
      uname_machine = os.uname().machine
  
      output = f"""
  Use thread_id={context.bash_state.current_thread_id} for all wcgw tool calls which take that.
  ---
  {mode_prompt}
  
  # Environment
  System: {uname_sysname}
  Machine: {uname_machine}
  Initialized in directory (also cwd): {context.bash_state.cwd}
  User home directory: {expanduser("~")}
  
  {alignment_context}
  {repo_context}
  
  ---
  
  {memory}
  ---
  
  {initial_files_context}
  
  """
  
      return output, context, initial_paths_with_ranges

• src/wcgw/types_.py:51-93 (schema)

  Pydantic BaseModel defining the input schema for the Initialize tool, including fields for initialization type, workspace path, files to read, task resumption, mode, thread ID, and code writer config.

  class Initialize(BaseModel):
      type: Literal[
          "first_call",
          "user_asked_mode_change",
          "reset_shell",
          "user_asked_change_workspace",
      ]
      any_workspace_path: str = Field(
          description="Workspace to initialise in. Don't use ~ by default, instead use empty string"
      )
      initial_files_to_read: list[str] = Field(
          description="Array of one or more files to read. Provide [] if no files mentioned."
      )
      task_id_to_resume: str
      mode_name: Literal["wcgw", "architect", "code_writer"]
      thread_id: str = Field(
          description="Use the thread_id created in first_call, leave it as empty string if first_call"
      )
      code_writer_config: Optional[CodeWriterMode] = None
  
      def model_post_init(self, __context: Any) -> None:
          self.thread_id = normalize_thread_id(self.thread_id)
          if self.mode_name == "code_writer":
              assert self.code_writer_config is not None, (
                  "code_writer_config can't be null when the mode is code_writer"
              )
          if self.type != "first_call" and not self.thread_id:
              raise ValueError(
                  "Thread id should be provided if type != 'first_call', including when resetting"
              )
          return super().model_post_init(__context)
  
      @property
      def mode(self) -> ModesConfig:
          if self.mode_name == "wcgw":
              return "wcgw"
          if self.mode_name == "architect":
              return "architect"
          assert self.code_writer_config is not None, (
              "code_writer_config can't be null when the mode is code_writer"
          )
          return self.code_writer_config

• src/wcgw/client/tool_prompts.py:19-36 (registration)

  MCP Tool registration for Initialize, providing the input schema from the Pydantic model, detailed description of usage, and annotations indicating it's read-only.

  TOOL_PROMPTS = [
      Tool(
          inputSchema=remove_titles_from_schema(Initialize.model_json_schema()),
          name="Initialize",
          description="""
  - Always call this at the start of the conversation before using any of the shell tools from wcgw.
  - Use `any_workspace_path` to initialize the shell in the appropriate project directory.
  - If the user has mentioned a workspace or project root or any other file or folder use it to set `any_workspace_path`.
  - If user has mentioned any files use `initial_files_to_read` to read, use absolute paths only (~ allowed)
  - By default use mode "wcgw"
  - In "code-writer" mode, set the commands and globs which user asked to set, otherwise use 'all'.
  - Use type="first_call" if it's the first call to this tool.
  - Use type="user_asked_mode_change" if in a conversation user has asked to change mode.
  - Use type="reset_shell" if in a conversation shell is not working after multiple tries.
  - Use type="user_asked_change_workspace" if in a conversation user asked to change workspace
  """,
          annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False),
      ),

• src/wcgw/client/tools.py:1040-1082 (handler)

  Dispatch handler in get_tool_output that routes Initialize calls to the main initialize function or reset_wcgw based on type, and handles file path tracking for whitelist.

  context.console.print("Calling initial info tool")
  if arg.type == "user_asked_mode_change" or arg.type == "reset_shell":
      workspace_path = (
          arg.any_workspace_path
          if os.path.isdir(arg.any_workspace_path)
          else os.path.dirname(arg.any_workspace_path)
      )
      workspace_path = workspace_path if os.path.exists(workspace_path) else ""
  
      # For these specific operations, thread_id is required
      output = (
          reset_wcgw(
              context,
              workspace_path,
              arg.mode_name
              if is_mode_change(arg.mode, context.bash_state)
              else None,
              arg.mode,
              arg.thread_id,
          ),
          0.0,
      )
  else:
      output_, context, init_paths = initialize(
          arg.type,
          context,
          arg.any_workspace_path,
          arg.initial_files_to_read or [],
          arg.task_id_to_resume,
          coding_max_tokens,
          noncoding_max_tokens,
          arg.mode,
          arg.thread_id,
      )
      output = output_, 0.0
      # Since init_paths is already a dictionary mapping file paths to line ranges,
      # we just need to merge it with our tracking dictionary
      for path, ranges in init_paths.items():
          if path not in file_paths_with_ranges and os.path.exists(path):
              file_paths_with_ranges[path] = ranges
          elif path in file_paths_with_ranges:
              file_paths_with_ranges[path].extend(ranges)

• src/wcgw/client/tools.py:912-927 (helper)

  which_tool_name helper maps tool name 'Initialize' to the Initialize Pydantic class.

  def which_tool_name(name: str) -> Type[TOOLS]:
      if name == "BashCommand":
          return BashCommand
      elif name == "FileWriteOrEdit":
          return FileWriteOrEdit
      elif name == "ReadImage":
          return ReadImage
      elif name == "ReadFiles":
          return ReadFiles
      elif name == "Initialize":
          return Initialize
      elif name == "ContextSave":
          return ContextSave
      else:
          raise ValueError(f"Unknown tool name: {name}")