query_measurements

Query network performance measurements from perfSONAR archives to analyze throughput, latency, and packet loss data with customizable filters.

Instructions

Query perfSONAR measurements with optional filters. Returns metadata about available measurements.

Input Schema

TableJSON Schema

Name	Required	Description
`source`	No	Source host/IP address
`destination`	No	Destination host/IP address
`eventType`	No	Event type to filter
`toolName`	No	Tool name to filter
`timeRange`	No	Time range in seconds

Implementation Reference

src/perfsonar_mcp/fastmcp_server.py:94-121 (handler)

The handler function `query_measurements` which is exposed as an MCP tool via the fastmcp framework, mapping user inputs to `MeasurementQueryParams` and delegating the actual network call to `perfsonar_client.query_measurements`.

async def query_measurements(
    source: Optional[str] = None,
    destination: Optional[str] = None,
    eventType: Optional[str] = None,
    toolName: Optional[str] = None,
    timeRange: Optional[int] = None,
) -> str:
    """Query perfSONAR measurements with optional filters. Returns metadata about available measurements.

    Args:
        source: Source host/IP address
        destination: Destination host/IP address
        eventType: Event type to filter (e.g., 'throughput', 'histogram-owdelay')
        toolName: Tool name to filter (e.g., 'bwctl/iperf3', 'powstream')
        timeRange: Time range in seconds from now

    Returns:
        JSON string with measurement metadata
    """
    params = MeasurementQueryParams(
        source=source,
        destination=destination,
        event_type=eventType,
        tool_name=toolName,
        time_range=timeRange,
    )
    results = await perfsonar_client.query_measurements(params)
    return json.dumps([r.model_dump(by_alias=True) for r in results], indent=2)

src/perfsonar_mcp/client.py:40-66 (handler)

The underlying client method `query_measurements` that performs the actual HTTP request to the perfSONAR service.

async def query_measurements(
    self, params: Optional[MeasurementQueryParams] = None
) -> List[MeasurementMetadata]:
    """
    Query measurements with optional filters

    Args:
        params: Query parameters for filtering measurements

    Returns:
        List of measurement metadata
    """
    logger.info("Querying measurements")
    logger.debug(f"Query parameters: {params}")
    try:
        query_params = {}
        if params:
            query_params = params.model_dump(by_alias=True, exclude_none=True)

        response = await self.client.get("/", params=query_params)
        response.raise_for_status()

        data = response.json()
        logger.info(f"Retrieved {len(data)} measurement records")
        return [MeasurementMetadata.model_validate(item) for item in data]
    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP error querying measurements: {e.response.status_code}")

perfsonar-mcp