perfsonar-mcp

Instructions

Implementation Reference

• src/perfsonar_mcp/pscheduler.py:180-210 (handler)

  The core implementation of the RTT test scheduling logic, which interacts with the pScheduler service.

  async def schedule_rtt_test(
      self, dest: str, count: int = 10, slip: str = "PT10M"
  ) -> PSchedulerTaskResponse:
      """
      Schedule an RTT (round-trip time) test
  
      Args:
          dest: Destination host
          count: Number of pings
          slip: Schedule slip time in ISO 8601 format (e.g., PT10M for 10 minutes)
  
      Returns:
          Task response
      """
      logger.info(f"Scheduling RTT test to {dest} ({count} pings)")
  
      # For RTT tests, schedule on the destination since it measures from local to dest
      scheduler_url = f"https://{dest}/pscheduler"
      logger.info(f"Using pScheduler at: {scheduler_url}")
  
      test_spec = RTTTestSpec(dest=dest, count=count)
  
      task_request = PSchedulerTaskRequest(
          test=PSchedulerTestSpec(type="rtt", spec=test_spec.model_dump(exclude_none=True)),
          schedule={"slip": slip},
      )
  
      # Create a temporary client for this specific scheduler
      client = PSchedulerClient(scheduler_url)
      try:
          return await client.create_task(task_request)

• src/perfsonar_mcp/fastmcp_server.py:332-348 (handler)

  The MCP-exposed wrapper function for scheduling RTT tests.

  async def schedule_rtt_test(
      dest: str,
      count: int = 10,
      slip: str = "PT10M",
  ) -> str:
      """Schedule an RTT (ping) test using pScheduler.
  
      Args:
          dest: Destination host for the test
          count: Number of pings (default: 10)
          slip: Schedule slip time in ISO 8601 format (e.g., 'PT10M' for 10 minutes)
  
      Returns:
          JSON string with test details including run URL for status checks
      """
      result = await pscheduler_client.schedule_rtt_test(dest, count, slip)
      return json.dumps(result.model_dump(), indent=2)

• src/perfsonar_mcp/server.py:234-240 (registration)

  Registration of the schedule_rtt_test tool in the MCP server.

  Tool(
      name="schedule_rtt_test",
      description="Schedule an RTT (ping) test using pScheduler.",
      inputSchema={
          "type": "object",
          "properties": {
              "dest": {"type": "string", "description": "Destination host"},