mcp-wcgw

Instructions

Implementation Reference

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

  The core handler function for the 'Initialize' tool. It sets up the workspace, loads or creates bash state, handles task resumption, reads initial files, generates repo context and mode prompts, and returns initialization output including context and file ranges.
  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:45-84 (schema)

  Pydantic BaseModel defining the input schema and validation for the 'Initialize' tool arguments.
  class Initialize(BaseModel): type: Literal[ "first_call", "user_asked_mode_change", "reset_shell", "user_asked_change_workspace", ] any_workspace_path: str = Field( description="Workspce to initialise in. Don't use ~ by default, instead use empty string" ) initial_files_to_read: list[str] 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: 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:20-36 (registration)

  MCP tool registration: defines the 'Initialize' tool with its JSON schema (derived from Initialize model), name, description, and annotations. This list is returned by server.list_tools().
  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/mcp_server/server.py:94-148 (registration)

  MCP server handler for tool calls. Dispatches any tool call (including 'Initialize') by parsing arguments into the tool model using which_tool_name and parse_tool_by_name, then calls get_tool_output which invokes the specific handler.
  @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/client/tools.py:1040-1082 (helper)

  Dispatch logic within get_tool_output that recognizes Initialize tool args and calls the initialize handler function.
  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, 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)