get_latency
Measure network latency between source and destination hosts to identify performance issues and optimize connectivity using historical delay data.
Instructions
Get latency/delay measurements between source and destination.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| source | Yes | Source host/IP address | |
| destination | Yes | Destination host/IP address | |
| timeRange | No | Time range in seconds | |
| summaryWindow | No | Summary window in seconds |
Implementation Reference
- 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