AgentSkills MCP

Instructions

Implementation Reference

• agentskills_mcp/core/tools/run_shell_command_op.py:103-185 (handler)

  Implements the core execution logic for the 'run_shell_command' tool: parses input, optionally installs Python dependencies using pipreqs, executes the command in the skill's directory via subprocess, captures and combines stdout/stderr as output.

  async def async_execute(self):
      """Execute the shell command operation.
  
      Executes a shell command in the specified skill's directory. For Python
      commands, the method attempts to automatically detect and install
      dependencies using pipreqs before executing the command.
  
      The method:
      1. Extracts skill_name and command from input_dict
      2. Looks up the skill directory from skill_metadata_dict
      3. For Python commands (containing "py"), checks if pipreqs is available
      4. If pipreqs is available, generates requirements.txt and installs dependencies
      5. Constructs the full command as `cd {skill_dir} && {command}`
      6. Executes the command in a subprocess with the current environment
      7. Captures stdout and stderr output
      8. Returns the combined output (stdout + stderr)
  
      Returns:
          None: The result is set via `self.set_output()` with the combined
              stdout and stderr output from the command execution. The output
              is decoded as UTF-8 and formatted as: "{stdout}\n{stderr}"
  
      Raises:
          KeyError: If skill_name is not found in skill_metadata_dict.
              This should be handled by ensuring LoadSkillMetadataOp is
              called before RunShellCommandOp.
  
      Note:
          - Dependency auto-installation only occurs for commands containing "py"
            and when the auto_install_deps parameter is enabled
          - If pipreqs is not available, a warning is logged but execution continues
          - If dependency installation fails, a warning is logged but the command
            is still executed
          - The command runs in the skill's directory, allowing access to
            skill-specific files and resources
          - Environment variables from the current process are passed to the subprocess
      """
      # Extract skill name and command from input parameters
      skill_name = self.input_dict["skill_name"]
      command: str = self.input_dict["command"]
      # Look up the skill directory from the metadata dictionary
      # This dictionary should be populated by LoadSkillMetadataOp
      # skill_dir = Path(self.context.skill_metadata_dict[skill_name]["skill_dir"])
      skill_dir = Path(C.service_config.metadata["skill_dir"])
      logger.info(f"🔧 run shell command: skill_name={skill_name} skill_dir={skill_dir} command={command}")
  
      # Auto-install dependencies for Python scripts if pipreqs is available
      # This helps ensure that Python scripts have their required dependencies
      # Only install if auto_install_deps parameter is enabled
      if self.auto_install_deps:
          if "py" in command:
              pipreqs_available = shutil.which("pipreqs") is not None
              if pipreqs_available:
                  install_cmd = f"cd {skill_dir}/{skill_name} && pipreqs . --force && pip install -r requirements.txt"
                  proc = await asyncio.create_subprocess_shell(
                      install_cmd,
                      stdout=asyncio.subprocess.PIPE,
                      stderr=asyncio.subprocess.PIPE,
                  )
                  stdout, stderr = await proc.communicate()
                  if proc.returncode != 0:
                      logger.warning(f"⚠️ Failed to install dependencies:\n{stdout.decode()}\n{stderr.decode()}")
                  else:
                      logger.info(f"✅ Dependencies installed successfully.\n{stdout.decode()}\n{stderr.decode()}")
              else:
                  logger.info("❗️ pipreqs not found, skipping dependency auto-install.")
  
      # Construct the full command to execute in the skill directory
      # This ensures the command runs in the correct context
      full_command = f"cd {skill_dir}/{skill_name} && {command}"
      proc = await asyncio.create_subprocess_shell(
          full_command,
          stdout=asyncio.subprocess.PIPE,
          stderr=asyncio.subprocess.PIPE,
          env=os.environ.copy(),
      )
  
      # Wait for the command to complete and capture output
      stdout, stderr = await proc.communicate()
      # Combine stdout and stderr output, decoded as UTF-8
      output = stdout.decode().strip() + "\n" + stderr.decode().strip()
      logger.info(f"✅ Command executed: skill_name={skill_name} output={output}")
      self.set_output(output)

• agentskills_mcp/core/tools/run_shell_command_op.py:84-101 (schema)

  Defines the ToolCall schema for the 'run_shell_command' tool, including name, description, and required input parameters 'skill_name' (string) and 'command' (string).

  return ToolCall(
      **{
          "name": "run_shell_command",
          "description": self.get_prompt("tool_desc"),
          "input_schema": {
              "skill_name": {
                  "type": "string",
                  "description": "skill name",
                  "required": True,
              },
              "command": {
                  "type": "string",
                  "description": "shell command",
                  "required": True,
              },
          },
      },
  )

• agentskills_mcp/core/tools/run_shell_command_op.py:22-23 (registration)

  Registers the RunShellCommandOp class as an operation using the @C.register_op() decorator from FlowLLM, which exposes it as the 'run_shell_command' tool in the MCP server.

  @C.register_op()
  class RunShellCommandOp(BaseAsyncToolOp):

• agentskills_mcp/core/tools/__init__.py:13-13 (registration)

  Imports RunShellCommandOp into the tools package __init__.py, making it available for use in the broader application.

  from .run_shell_command_op import RunShellCommandOp