perfsonar-mcp

Instructions

Implementation Reference

• src/perfsonar_mcp/pscheduler.py:132-178 (handler)

  The actual implementation of scheduling a latency test in the PSchedulerClient class.

  async def schedule_latency_test(
      self,
      source: Optional[str],
      dest: str,
      packet_count: int = 600,
      packet_interval: float = 0.1,
      slip: str = "PT10M",
  ) -> PSchedulerTaskResponse:
      """
      Schedule a latency (one-way delay) test
  
      Args:
          source: Source host (None for local)
          dest: Destination host
          packet_count: Number of packets to send
          packet_interval: Interval between packets in seconds
          slip: Schedule slip time in ISO 8601 format (e.g., PT10M for 10 minutes)
  
      Returns:
          Task response
      """
      logger.info(
          f"Scheduling latency test from {source or 'local'} to {dest} ({packet_count} packets)"
      )
  
      # Determine which node to schedule on (prefer source if available)
      scheduler_node = source or dest
      scheduler_url = f"https://{scheduler_node}/pscheduler"
      logger.info(f"Using pScheduler at: {scheduler_url}")
  
      test_spec = LatencyTestSpec(
          source=source, dest=dest, packet_count=packet_count, packet_interval=packet_interval
      )
  
      task_request = PSchedulerTaskRequest(
          test=PSchedulerTestSpec(
              type="latency", spec=test_spec.model_dump(by_alias=True, 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)
      finally:
          await client.close()