sipp_run_scenario

Instructions

Implementation Reference

• src/opensips_mcp/tools/ecosystem_tools.py:119-146 (handler)

  MCP tool handler for 'sipp_run_scenario'. Decorated with @mcp.tool(), @audited('sipp_run_scenario'), and @require_permission('process.manage'). Parses optional JSON options and delegates to SIPpRunner.run_scenario().

  @mcp.tool()
  @audited("sipp_run_scenario")
  @require_permission("process.manage")
  async def sipp_run_scenario(
      ctx: Context,
      scenario_file: str,
      target: str,
      options: str | None = None,
  ) -> dict[str, Any]:
      """Run a SIPp XML scenario file against a target SIP server.
  
      Parameters
      ----------
      scenario_file:
          Path to the .xml SIPp scenario file.
      target:
          Target host:port (e.g. ``192.168.1.1:5060``).
      options:
          JSON string of extra CLI flags, e.g. ``{"-m": "10", "-r": "5"}``.
      """
      opts: dict[str, str] | None = None
      if options:
          try:
              opts = json.loads(options)
          except json.JSONDecodeError:
              return {"error": f"Invalid options JSON: {options}"}
  
      return await SIPpRunner.run_scenario(scenario_file, target, options=opts)

• src/opensips_mcp/ecosystem/sipp.py:16-69 (helper)

  SIPpRunner.run_scenario() static method — the actual implementation that validates the scenario file exists, builds the 'sipp' CLI command (with -sf, -trace_err, -trace_stat), launches it as a subprocess, and returns stdout/stderr/return_code.

  @staticmethod
  async def run_scenario(
      scenario_file: str,
      target: str,
      *,
      options: dict[str, str] | None = None,
      timeout: int = 120,
  ) -> dict[str, Any]:
      """Run a SIPp XML scenario against a target.
  
      Parameters
      ----------
      scenario_file:
          Path to the ``.xml`` SIPp scenario file.
      target:
          Target host (``host:port`` format).
      options:
          Extra CLI flags as key/value pairs (e.g. ``{"-m": "10"}``).
      timeout:
          Max seconds to wait for the process.
      """
      if not Path(scenario_file).is_file():
          return {"error": f"Scenario file not found: {scenario_file}"}
  
      cmd: list[str] = [
          "sipp",
          target,
          "-sf", scenario_file,
          "-trace_err",
          "-trace_stat",
      ]
  
      if options:
          for flag, value in options.items():
              cmd.extend([flag, value])
  
      try:
          proc = await asyncio.create_subprocess_exec(
              *cmd,
              stdout=asyncio.subprocess.PIPE,
              stderr=asyncio.subprocess.PIPE,
          )
          stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
          return {
              "return_code": proc.returncode,
              "stdout": stdout.decode(errors="replace")[-4096:],
              "stderr": stderr.decode(errors="replace")[-2048:],
              "scenario": scenario_file,
              "target": target,
          }
      except asyncio.TimeoutError:
          return {"error": "SIPp process timed out", "timeout": timeout}
      except FileNotFoundError:
          return {"error": "sipp binary not found — install SIPp first"}

• src/opensips_mcp/resources/ecosystem_resources.py:30-35 (schema)

  Resource descriptor for 'sipp_run_scenario' documenting parameters (scenario_file, target, options), required permission (process.manage), and an example invocation.

  "sipp_run_scenario": {
      "description": "Run a SIPp XML scenario against a target",
      "parameters": "scenario_file, target, options (JSON string)",
      "permission": "process.manage",
      "example": 'sipp_run_scenario(scenario_file="/etc/sipp/uac.xml", target="10.0.0.1:5060")',
  },

• src/opensips_mcp/tools/ecosystem_tools.py:424-424 (registration)

  Listing of 'sipp_run_scenario' under the 'sipp' tool group in the ecosystem_health overview.

  "sipp": ["sipp_run_scenario", "sipp_load_test", "sipp_list_scenarios"],