repo_description: Shell and coding agent on mcp clients
repo_topics: agent ai-agent ai-coding anthropic chatgpt claude claude-desktop custom-gpt llm llm-agent llm-code mcp mcp-server openai shell terminal
code: import os
  import re
  from typing import Any, List, Literal, Optional, Protocol, Sequence, Union
  
  from pydantic import BaseModel as PydanticBaseModel
  from pydantic import Field, PrivateAttr, model_serializer, model_validator
  
  
  _PIPE_TAIL_RE = re.compile(r"\|\s*tail(?:\s+(?:-n\s*|-)?(\d+))?\s*$")
  
  
  def strip_tail_pipe(command: str) -> str:
      """Strip trailing `| tail ...` from a command.
  
      LLMs frequently pipe output through tail, but wcgw already manages output
      truncation server-side.  Stripping the pipe avoids hiding useful earlier
      output from the model.
      """
      match = _PIPE_TAIL_RE.search(command)
      if match:
          return command[: match.start()].rstrip()
      return command
  
  
  def normalize_thread_id(thread_id: str) -> str:
      """Normalize thread_id by keeping only word characters (alphanumeric and underscore)."""
      return re.sub(r"[^\w]", "", thread_id)
  
  
  class NoExtraArgs(PydanticBaseModel):
      class Config:
          extra = "forbid"
  
  
  BaseModel = NoExtraArgs
  
  
  Modes = Literal["wcgw", "architect", "code_writer"]
  
  
  class CodeWriterMode(BaseModel):
      allowed_globs: Literal["all"] | list[str]
      allowed_commands: Literal["all"] | list[str]
  
      def model_post_init(self, _: Any) -> None:
          # Patch frequently wrong output trading off accuracy
          # in rare case there's a file named 'all' or a command named 'all'
          if isinstance(self.allowed_commands, list) and len(self.allowed_commands) == 1:
              if self.allowed_commands[0] == "all":
                  self.allowed_commands = "all"
          if isinstance(self.allowed_globs, list) and len(self.allowed_globs) == 1:
              if self.allowed_globs[0] == "all":
                  self.allowed_globs = "all"
  
      def update_relative_globs(self, workspace_root: str) -> None:
          """Update globs if they're relative paths"""
          if self.allowed_globs != "all":
              self.allowed_globs = [
                  glob if os.path.isabs(glob) else os.path.join(workspace_root, glob)
                  for glob in self.allowed_globs
              ]
  
  
  ModesConfig = Union[Literal["wcgw", "architect"], CodeWriterMode]
  
  
  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"
      )
      allowed_globs: Optional[Literal["all"] | list[str]] = Field(
          default=None,
          description="File globs that are allowed to be edited. Set to 'all' to allow all files, or provide a list of glob patterns. Only required when mode_name is 'code_writer'.",
      )
      allowed_commands: Optional[Literal["all"] | list[str]] = Field(
          default=None,
          description="Shell commands that are allowed to be executed. Set to 'all' to allow all commands, or provide a list of command patterns. Only required when mode_name is 'code_writer'.",
      )
  
      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.allowed_globs is not None, (
                  "allowed_globs can't be null when the mode is code_writer"
              )
              assert self.allowed_commands is not None, (
                  "allowed_commands can't be null when the mode is code_writer"
              )
              # Patch frequently wrong output trading off accuracy
              # in rare case there's a file named 'all' or a command named 'all'
              if (
                  isinstance(self.allowed_commands, list)
                  and len(self.allowed_commands) == 1
              ):
                  if self.allowed_commands[0] == "all":
                      self.allowed_commands = "all"
              if isinstance(self.allowed_globs, list) and len(self.allowed_globs) == 1:
                  if self.allowed_globs[0] == "all":
                      self.allowed_globs = "all"
          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.allowed_globs is not None, (
              "allowed_globs can't be null when the mode is code_writer"
          )
          assert self.allowed_commands is not None, (
              "allowed_commands can't be null when the mode is code_writer"
          )
          config = CodeWriterMode(
              allowed_globs=self.allowed_globs, allowed_commands=self.allowed_commands
          )
          return config
  
      def update_relative_globs(self, workspace_root: str) -> None:
          """Update globs if they're relative paths"""
          if self.allowed_globs is not None and self.allowed_globs != "all":
              self.allowed_globs = [
                  glob if os.path.isabs(glob) else os.path.join(workspace_root, glob)
                  for glob in self.allowed_globs
              ]
  
  
  class CommandBase(BaseModel):
      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)
  
  
  class Command(CommandBase):
      command: str
      type: Literal["command"] = "command"
      is_background: bool = False
  
      def model_post_init(self, __context: Any) -> None:
          self.command = strip_tail_pipe(self.command)
          return super().model_post_init(__context)
  
  
  class StatusCheck(CommandBase):
      status_check: Literal[True] = True
      type: Literal["status_check"] = "status_check"
      bg_command_id: str | None = None
  
  
  class SendText(CommandBase):
      send_text: str
      type: Literal["send_text"] = "send_text"
      bg_command_id: str | None = None
  
  
  Specials = Literal[
      "Enter", "Key-up", "Key-down", "Key-left", "Key-right", "Ctrl-c", "Ctrl-d"
  ]
  
  
  class SendSpecials(CommandBase):
      send_specials: Sequence[Specials]
      type: Literal["send_specials"] = "send_specials"
      bg_command_id: str | None = None
  
  
  class SendAscii(CommandBase):
      send_ascii: Sequence[int]
      type: Literal["send_ascii"] = "send_ascii"
      bg_command_id: str | None = None
  
  
  class ActionJsonSchema(BaseModel):
      type: Literal[
          "command", "status_check", "send_text", "send_specials", "send_ascii"
      ] = Field(description="type of action.")
      command: Optional[str] = Field(
          default=None, description='Set only if type="command"'
      )
      status_check: Optional[Literal[True]] = Field(
          default=None, description='Set only if type="status_check"'
      )
      send_text: Optional[str] = Field(
          default=None, description='Set only if type="send_text"'
      )
      send_specials: Optional[Sequence[Specials]] = Field(
          default=None, description='Set only if type="send_specials"'
      )
      send_ascii: Optional[Sequence[int]] = Field(
          default=None, description='Set only if type="send_ascii"'
      )
      is_background: bool = Field(
          default=False,
          description='Set only if type="command" and running the command in background',
      )
      bg_command_id: str | None = Field(
          default=None,
          description='Set only if type!="command" and doing action on a running background command',
      )
      wait_for_seconds: Optional[float] = None
      thread_id: str
  
  
  class BashCommand(BaseModel):
      action_json: Command | StatusCheck | SendText | SendSpecials | SendAscii
  
      @model_validator(mode="before")
      @classmethod
      def combine(cls, data: Any) -> Any:
          # If action_json is already provided, don't wrap it
          if isinstance(data, dict) and "action_json" in data:
              return data
          # Otherwise wrap the data in action_json
          return {"action_json": data}
  
      @model_serializer(mode="plain")
      def serialize_model(self) -> dict[str, Any]:
          return self.action_json.model_dump()
  
      @staticmethod
      def model_json_schema(*args, **kwargs) -> dict[str, Any]:  # type: ignore
          return ActionJsonSchema.model_json_schema(*args, **kwargs)
  
  
  class ReadImage(BaseModel):
      file_path: str
  
  
  class WriteIfEmpty(BaseModel):
      file_path: str
      file_content: str
  
  
  class ReadFiles(BaseModel):
      file_paths: list[str]
      _start_line_nums: List[Optional[int]] = PrivateAttr(default_factory=lambda: [])
      _end_line_nums: List[Optional[int]] = PrivateAttr(default_factory=lambda: [])
  
      @property
      def show_line_numbers_reason(self) -> str:
          return "True"
  
      @property
      def start_line_nums(self) -> List[Optional[int]]:
          """Get the start line numbers."""
          return self._start_line_nums
  
      @property
      def end_line_nums(self) -> List[Optional[int]]:
          """Get the end line numbers."""
          return self._end_line_nums
  
      def model_post_init(self, __context: Any) -> None:
          # Parse file paths for line ranges and store them in private attributes
          self._start_line_nums = []
          self._end_line_nums = []
  
          # Create new file_paths list without line ranges
          clean_file_paths = []
  
          for file_path in self.file_paths:
              start_line_num = None
              end_line_num = None
              path_part = file_path
  
              # Check if the path ends with a line range pattern
              # We're looking for patterns at the very end of the path like:
              #  - file.py:10      (specific line)
              #  - file.py:10-20   (line range)
              #  - file.py:10-     (from line 10 to end)
              #  - file.py:-20     (from start to line 20)
  
              # Split by the last colon
              if ":" in file_path:
                  parts = file_path.rsplit(":", 1)
                  if len(parts) == 2:
                      potential_path = parts[0]
                      line_spec = parts[1]
  
                      # Check if it's a valid line range format
                      if line_spec.isdigit():
                          # Format: file.py:10
                          try:
                              start_line_num = int(line_spec)
                              path_part = potential_path
                          except ValueError:
                              # Keep the original path if conversion fails
                              pass
  
                      elif "-" in line_spec:
                          # Could be file.py:10-20, file.py:10-, or file.py:-20
                          line_parts = line_spec.split("-", 1)
  
                          if not line_parts[0] and line_parts[1].isdigit():
                              # Format: file.py:-20
                              try:
                                  end_line_num = int(line_parts[1])
                                  path_part = potential_path
                              except ValueError:
                                  # Keep original path
                                  pass
  
                          elif line_parts[0].isdigit():
                              # Format: file.py:10-20 or file.py:10-
                              try:
                                  start_line_num = int(line_parts[0])
  
                                  if line_parts[1].isdigit():
                                      # file.py:10-20
                                      end_line_num = int(line_parts[1])
  
                                  # In both cases, update the path
                                  path_part = potential_path
                              except ValueError:
                                  # Keep original path
                                  pass
  
              # Add clean path and corresponding line numbers
              clean_file_paths.append(path_part)
              self._start_line_nums.append(start_line_num)
              self._end_line_nums.append(end_line_num)
  
          # Update file_paths with clean paths
          self.file_paths = clean_file_paths
  
          return super().model_post_init(__context)
  
  
  class FileEdit(BaseModel):
      file_path: str
      file_edit_using_search_replace_blocks: str
  
  
  class FileWriteOrEdit(BaseModel):
      # Naming should be in sorted order otherwise it gets changed in LLM backend.
      file_path: str = Field(description="#1: absolute file path")
      percentage_to_change: int = Field(
          description="#2: predict this percentage, calculated as number of existing lines that will have some diff divided by total existing lines."
      )
      text_or_search_replace_blocks: str = Field(
          description="#3: content/edit blocks. Must be after #2 in the tool xml"
      )
      thread_id: str = Field(description="#4: thread_id")
  
      def model_post_init(self, __context: Any) -> None:
          self.thread_id = normalize_thread_id(self.thread_id)
          return super().model_post_init(__context)
  
  
  class ContextSave(BaseModel):
      id: str
      project_root_path: str
      description: str
      relevant_file_globs: list[str]
  
  
  class Console(Protocol):
      def print(self, *objects: Any, **kwargs: Any) -> None: ...
  
      def log(self, *objects: Any, **kwargs: Any) -> None: ...
  
  
  class Mdata(PydanticBaseModel):
      data: BashCommand | FileWriteOrEdit | str | ReadFiles | Initialize | ContextSave
  