perfsonar-mcp

Instructions

Implementation Reference

• src/perfsonar_mcp/fastmcp_server.py:284-302 (handler)

  FastMCP tool definition for schedule_throughput_test, which calls the underlying pScheduler client.

  async def schedule_throughput_test(
      dest: str,
      source: Optional[str] = None,
      duration: str = "PT30S",
      slip: str = "PT10M",
  ) -> str:
      """Schedule a throughput test using pScheduler.
  
      Args:
          dest: Destination host for the test
          source: Optional source host (if not specified, uses pScheduler host)
          duration: Test duration in ISO 8601 format (e.g., 'PT30S' for 30 seconds)
          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_throughput_test(source, dest, duration, slip)
      return json.dumps(result.model_dump(), indent=2)

• src/perfsonar_mcp/pscheduler.py:88-130 (handler)

  Actual implementation of the throughput test scheduling logic in the pScheduler client.

  async def schedule_throughput_test(
      self,
      source: Optional[str],
      dest: str,
      duration: str = "PT30S",
      slip: str = "PT10M",
  ) -> PSchedulerTaskResponse:
      """
      Schedule a throughput test
  
      Args:
          source: Source host (None for local)
          dest: Destination host
          duration: Test duration in ISO 8601 format (e.g., PT30S for 30 seconds)
          slip: Schedule slip time in ISO 8601 format (e.g., PT10M for 10 minutes)
  
      Returns:
          Task response
      """
      logger.info(
          f"Scheduling throughput test from {source or 'local'} to {dest} with duration {duration}"
      )
  
      # 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 = ThroughputTestSpec(source=source, dest=dest, duration=duration)
  
      task_request = PSchedulerTaskRequest(
          test=PSchedulerTestSpec(
              type="throughput", 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)
      finally:
          await client.close()