get_throughput
Measure network throughput between source and destination hosts using historical performance data from perfSONAR archives.
Instructions
Get throughput 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 MCP tool handler for 'get_throughput' which calls the perfsonar_client and formats the result as a JSON string.
async def get_throughput( source: str, destination: str, timeRange: int = 86400, summaryWindow: Optional[int] = None, ) -> str: """Get throughput 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 throughput measurement data """ results = await perfsonar_client.get_throughput(source, destination, timeRange, summaryWindow) return json.dumps([r.model_dump(by_alias=True) for r in results], indent=2) - src/perfsonar_mcp/client.py:123-150 (handler)The underlying client implementation that fetches the actual throughput measurement data.
async def get_throughput( self, source: str, destination: str, time_range: Optional[int] = None, summary_window: Optional[int] = None, ) -> List[MeasurementResult]: """ Get throughput 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 throughput: {source} -> {destination}") metadata = await self.query_measurements( MeasurementQueryParams(source=source, destination=destination, event_type="throughput") ) results = [] for meta in metadata: event_type = next((e for e in meta.event_types if e.event_type == "throughput"), None) if not event_type: