perfsonar-mcp

Instructions

Implementation Reference

• src/perfsonar_mcp/client.py:74-115 (handler)

  The actual implementation of fetching measurement data from the PerfSONAR API.

  async def get_measurement_data(
      self, params: MeasurementDataParams
  ) -> List[TimeSeriesDataPoint]:
      """
      Get measurement data for a specific event type
  
      Args:
          params: Parameters specifying which measurement data to retrieve
  
      Returns:
          List of time series data points
      """
      logger.info(f"Getting measurement data for event type: {params.event_type}")
      logger.debug(f"Measurement data parameters: {params}")
      try:
          # Build the URL path
          path = f"/{params.metadata_key}/{params.event_type}"
  
          if params.summary_type and params.summary_window:
              path += f"/{params.summary_type}/{params.summary_window}"
          else:
              path += "/base"
  
          logger.debug(f"Request path: {path}")
  
          # Build query params
          query_params: Dict[str, Any] = {}
          if params.time_start:
              query_params["time-start"] = params.time_start
          if params.time_end:
              query_params["time-end"] = params.time_end
          if params.time_range:
              query_params["time-range"] = params.time_range
  
          response = await self.client.get(path, params=query_params)
          response.raise_for_status()
  
          data = response.json()
          logger.info(f"Retrieved {len(data)} data points")
          return [TimeSeriesDataPoint.model_validate(item) for item in data]
      except httpx.HTTPStatusError as e:
          logger.error(f"HTTP error getting measurement data: {e.response.status_code}")

• src/perfsonar_mcp/fastmcp_server.py:125-151 (handler)

  The MCP tool handler that defines the 'get_measurement_data' tool and calls the client implementation.

  async def get_measurement_data(
      metadataKey: str,
      eventType: str,
      summaryType: Optional[str] = None,
      summaryWindow: Optional[int] = None,
      timeRange: Optional[int] = None,
  ) -> str:
      """Get raw time-series data for a specific measurement.
  
      Args:
          metadataKey: Metadata key from query results
          eventType: Event type (e.g., 'throughput', 'histogram-owdelay')
          summaryType: Summary type (e.g., 'average', 'aggregation')
          summaryWindow: Summary window in seconds
          timeRange: Time range in seconds from now
  
      Returns:
          JSON string with time-series measurement data
      """
      params = MeasurementDataParams(
          metadata_key=metadataKey,
          event_type=eventType,
          summary_type=summaryType,
          summary_window=summaryWindow,
          time_range=timeRange,
      )
      results = await perfsonar_client.get_measurement_data(params)

• src/perfsonar_mcp/server.py:301-317 (registration)

  Where the 'get_measurement_data' tool is registered and handled in the generic MCP server implementation.

  elif name == "get_measurement_data":
      params = MeasurementDataParams(
          metadata_key=arguments["metadataKey"],
          event_type=arguments["eventType"],
          summary_type=arguments.get("summaryType"),
          summary_window=arguments.get("summaryWindow"),
          time_range=arguments.get("timeRange"),
      )
      results = await self.client.get_measurement_data(params)
      return CallToolResult(
          content=[
              TextContent(
                  type="text",
                  text=json.dumps([r.model_dump() for r in results], indent=2),
              )
          ]
      )