perfsonar-mcp

Instructions

Implementation Reference

• src/perfsonar_mcp/fastmcp_server.py:178-196 (handler)

  The FastMCP tool definition for `get_latency` which serves as the entry point for MCP requests.

  async def get_latency(
      source: str,
      destination: str,
      timeRange: int = 86400,
      summaryWindow: Optional[int] = None,
  ) -> str:
      """Get latency/delay measurements between source and destination.
  
      Args:
          source: Source host/IP address
          destination: Destination host/IP address
          timeRange: Time range in seconds (default: 86400 = 24 hours)
          summaryWindow: Summary window in seconds for aggregation
  
      Returns:
          JSON string with latency measurement data
      """
      results = await perfsonar_client.get_latency(source, destination, timeRange, summaryWindow)
      return json.dumps([r.model_dump(by_alias=True) for r in results], indent=2)

• src/perfsonar_mcp/client.py:168-227 (handler)

  The actual logic implementation of `get_latency` within the `PerfSONARClient` class.

  async def get_latency(
      self,
      source: str,
      destination: str,
      time_range: Optional[int] = None,
      summary_window: Optional[int] = None,
  ) -> List[MeasurementResult]:
      """
      Get latency/delay measurements between source and destination
  
      Args:
          source: Source host/IP address
          destination: Destination host/IP address
          time_range: Time range in seconds from now
          summary_window: Summary window in seconds
  
      Returns:
          List of measurement results
      """
      logger.info(f"Getting latency: {source} -> {destination}")
      # Try histogram-owdelay first, fall back to histogram-rtt
      metadata = await self.query_measurements(
          MeasurementQueryParams(
              source=source, destination=destination, event_type="histogram-owdelay"
          )
      )
  
      if not metadata:
          logger.debug("No histogram-owdelay data, trying histogram-rtt")
          metadata = await self.query_measurements(
              MeasurementQueryParams(
                  source=source, destination=destination, event_type="histogram-rtt"
              )
          )
  
      results = []
      for meta in metadata:
          event_types = ["histogram-owdelay", "histogram-rtt"]
          for event_type_name in event_types:
              event_type = next(
                  (e for e in meta.event_types if e.event_type == event_type_name), None
              )
              if not event_type:
                  continue
  
              data = await self.get_measurement_data(
                  MeasurementDataParams(
                      metadata_key=meta.metadata_key,
                      event_type=event_type_name,
                      summary_type="statistics" if summary_window else None,
                      summary_window=summary_window,
                      time_range=time_range,
                  )
              )
  
              results.append(MeasurementResult(metadata=meta, data=data))
              break
  
      logger.info(f"Retrieved {len(results)} latency results")
      return results