MCP Server API Documentation
Introduction
The Chompy Model Context Protocol (MCP) API enables you to integrate network monitoring data with your AI systems, internal LLMs, or automation workflows. Query devices, analyze traffic flows, run synthetic tests, and get network health insights programmatically.
Table of Contents
- Getting Started
- Authentication
- API Reference
- Tool Reference
- Code Examples
- Error Handling
- Rate Limits
- Best Practices
Getting Started
Base URL
https://your-chompy-instance.com/api/mcp
Quick Start
- Obtain an API key from your Chompy administrator
- Make your first request:
curl -X POST https://your-chompy-instance.com/api/mcp/tools/network\_health\_summary \ -H "Authorization: Bearer chmp_your_api_key" \ -H "Content-Type: application/json"
Authentication
All API requests require authentication via Bearer token.
Request Header
Authorization: Bearer chmp_your_api_key
Permission Levels
| Level | Description | Available Tools |
|---|---|---|
read_only | View data only | get_devices, get_alerts, top_talkers, search_flows, flow_summary, get_synthetic_tests, get_synthetic_results, get_snmp_metrics, get_interface_utilization, network_health_summary, get_device_details |
read_write | View + execute tests + acknowledge alerts | All read_only tools + acknowledge_alert, run_ping_test, run_traceroute, run_http_test |
admin | Full access including raw queries | All tools + query_postgres, query_clickhouse |
API Reference
List Available Tools
Returns all available tools and whether your API key can access them.
GET /api/mcp/tools
Response:
\{ "tools": [ \{ "name": "get_devices", "permission_required": "read_only", "available": true \}, ... ], "your_permission": "read_only" \}
Execute a Tool
POST /api/mcp/tools/\{tool_name\}
Headers:
Authorization: Bearer chmp_your_api_key Content-Type: application/json
Body: Tool-specific parameters (see Tool Reference)
Response:
\{ "success": true, "data": \{ ... \} \}
Tool Reference
Device Management
get_devices
List network devices from inventory with optional filters.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
status | string | "all" | Filter: up, down, unknown, all |
vendor | string | - | Filter by vendor (cisco, arista, juniper, etc.) |
site | string | - | Filter by site/location name |
device_type | string | - | Filter by type (router, switch, firewall) |
limit | number | 50 | Maximum results |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/get\_devices \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{ "status": "up", "vendor": "cisco", "limit": 10 \}'
Response:
\{ "success": true, "data": \{ "devices": [ \{ "hostname": "core-router-01", "ip": "192.168.1.1", "vendor": "Cisco", "model": "ISR 4451", "type": "router", "site": "HQ-DC1", "status": "up", "last_seen": "2025-12-13T10:30:00Z" \} ], "count": 1 \} \}
get_device_details
Get detailed information about a specific device including interfaces.
Parameters:
| Name | Type | Required | Description |
|---|---|---|---|
hostname | string | Yes | Device hostname or IP address |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/get\_device\_details \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{"hostname": "core-router-01"\}'
Alerts
get_alerts
Get alerts from the monitoring system.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
severity | string | "all" | Filter: critical, high, medium, low, info, all |
status | string | "active" | Filter: active, acknowledged, resolved, all |
source | string | - | Filter by source (snmp, synthetic, flow) |
limit | number | 25 | Maximum results |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/get\_alerts \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{ "severity": "critical", "status": "active" \}'
acknowledge_alert
Acknowledge an active alert. Requires read_write permission.
Parameters:
| Name | Type | Required | Description |
|---|---|---|---|
alert_id | number | Yes | Alert instance ID |
note | string | No | Acknowledgment note |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/acknowledge\_alert \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{ "alert_id": 1234, "note": "Investigating - ticket #5678" \}'
NetFlow Analysis
top_talkers
Get top bandwidth consumers from NetFlow data.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
minutes | number | 60 | Look back period |
direction | string | "source" | source, destination, or both |
limit | number | 10 | Number of results |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/top\_talkers \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{ "minutes": 30, "direction": "source", "limit": 5 \}'
Response:
\{ "success": true, "data": [ \{ "ip": "192.168.10.50", "bytes": "2.4 GB", "packets": 1850000, "flows": 12500 \} ] \}
search_flows
Search NetFlow data by IP, port, or protocol.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
src_ip | string | - | Source IP or CIDR (e.g., 192.168.1.0/24) |
dst_ip | string | - | Destination IP or CIDR |
src_port | number | - | Source port |
dst_port | number | - | Destination port |
protocol | string | - | tcp, udp, icmp, or protocol number |
minutes | number | 60 | Look back period |
limit | number | 100 | Maximum results |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/search\_flows \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{ "src_ip": "192.168.1.0/24", "dst_port": 443, "protocol": "tcp", "minutes": 15 \}'
flow_summary
Get traffic summary statistics over a time period.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
minutes | number | 60 | Look back period |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/flow\_summary \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{"minutes": 60\}'
Response:
\{ "success": true, "data": \{ "total_bytes": "45.2 GB", "total_packets": 38500000, "total_flows": 285000, "unique_sources": 1250, "unique_destinations": 3400 \} \}
Synthetic Monitoring
get_synthetic_tests
List configured synthetic monitoring tests.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
type | string | "all" | Filter: icmp, http, all |
enabled | boolean | - | Filter by enabled status |
get_synthetic_results
Get results from synthetic monitoring tests.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
test_type | string | "all" | Filter: icmp, http, all |
minutes | number | 60 | Look back period |
limit | number | 50 | Maximum results |
run_ping_test
Run an ad-hoc ICMP ping test. Requires read_write permission.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
target | string | Required | IP address or hostname |
count | number | 5 | Number of packets |
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/run\_ping\_test \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json" \ -d '\{ "target": "8.8.8.8", "count": 10 \}'
Response:
\{ "success": true, "data": \{ "target": "8.8.8.8", "packets_sent": 10, "packets_received": 10, "packet_loss": 0, "min_latency_ms": 12.5, "avg_latency_ms": 14.2, "max_latency_ms": 18.7 \} \}
run_traceroute
Run an ad-hoc traceroute. Requires read_write permission.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
target | string | Required | IP address or hostname |
max_hops | number | 30 | Maximum hops |
run_http_test
Run an ad-hoc HTTP test. Requires read_write permission.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
url | string | Required | Full URL including protocol |
method | string | "GET" | GET, POST, or HEAD |
timeout_seconds | number | 30 | Request timeout |
SNMP Metrics
get_snmp_metrics
Get recent SNMP polling metrics for devices.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
hostname | string | - | Filter by device hostname |
minutes | number | 60 | Look back period |
limit | number | 100 | Maximum results |
get_interface_utilization
Get interface utilization metrics for a device.
Parameters:
| Name | Type | Default | Description |
|---|---|---|---|
hostname | string | Required | Device hostname |
interface_name | string | - | Specific interface |
minutes | number | 60 | Look back period |
Network Health
network_health_summary
Get an overall network health summary.
Parameters: None
Example:
curl -X POST https://chompy.example.com/api/mcp/tools/network\_health\_summary \ -H "Authorization: Bearer chmp_xxx" \ -H "Content-Type: application/json"
Response:
\{ "success": true, "data": \{ "devices": [ \{"status": "up", "count": 45\}, \{"status": "down", "count": 2\} ], "active_alerts": [ \{"severity": "critical", "count": 1\}, \{"severity": "high", "count": 3\} ], "flow_stats": \{ "flows_last_hour": 285000, "bytes_last_hour": "45.2 GB" \} \} \}
Code Examples
Python
import requests
from typing import Optional, Dict, Any
class ChompyClient:
"""Chompy MCP API Client"""
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url.rstrip('/')
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def call_tool(self, tool_name: str, params: Optional[Dict] = None) -> Dict[str, Any]:
"""Execute an MCP tool"""
response = requests.post(
f"{self.base_url}/api/mcp/tools/{tool_name}",
headers=self.headers,
json=params or {}
)
response.raise_for_status()
return response.json()
def get_devices(self, status: str = "all", vendor: str = None, limit: int = 50):
"""Get network devices"""
params = {"status": status, "limit": limit}
if vendor:
params["vendor"] = vendor
return self.call_tool("get_devices", params)
def get_alerts(self, severity: str = "all", status: str = "active"):
"""Get alerts"""
return self.call_tool("get_alerts", {"severity": severity, "status": status})
def network_health(self):
"""Get network health summary"""
return self.call_tool("network_health_summary")
def top_talkers(self, minutes: int = 60, limit: int = 10):
"""Get top bandwidth consumers"""
return self.call_tool("top_talkers", {"minutes": minutes, "limit": limit})
def search_flows(self, src_ip: str = None, dst_ip: str = None,
dst_port: int = None, protocol: str = None, minutes: int = 60):
"""Search NetFlow data"""
params = {"minutes": minutes}
if src_ip: params["src_ip"] = src_ip
if dst_ip: params["dst_ip"] = dst_ip
if dst_port: params["dst_port"] = dst_port
if protocol: params["protocol"] = protocol
return self.call_tool("search_flows", params)
def run_ping(self, target: str, count: int = 5):
"""Run ad-hoc ping test"""
return self.call_tool("run_ping_test", {"target": target, "count": count})
# Usage Example
if __name__ == "__main__":
client = ChompyClient(
base_url="https://chompy.example.com",
api_key="chmp_your_api_key_here"
)
# Get network health
health = client.network_health()
print(f"Network Health: {health}")
# Get active critical alerts
alerts = client.get_alerts(severity="critical", status="active")
print(f"Critical Alerts: {alerts['data']['count']}")
# Find top talkers
talkers = client.top_talkers(minutes=30, limit=5)
for t in talkers['data']:
print(f" {t['ip']}: {t['bytes']}")
# Search for HTTPS traffic from a subnet
flows = client.search_flows(src_ip="192.168.1.0/24", dst_port=443, protocol="tcp")
print(f"HTTPS flows found: {len(flows['data'])}")
JavaScript / Node.js
const axios = require('axios');
class ChompyClient {
constructor(baseUrl, apiKey) {
this.baseUrl = baseUrl.replace(/\/$/, '');
this.headers = {
'Authorization': `Bearer ${apiKey}`,
'Content-Type': 'application/json'
};
}
async callTool(toolName, params = {}) {
const response = await axios.post(
`${this.baseUrl}/api/mcp/tools/${toolName}`,
params,
{ headers: this.headers }
);
return response.data;
}
async getDevices(options = {}) {
return this.callTool('get_devices', {
status: options.status || 'all',
vendor: options.vendor,
limit: options.limit || 50
});
}
async getAlerts(options = {}) {
return this.callTool('get_alerts', {
severity: options.severity || 'all',
status: options.status || 'active'
});
}
async networkHealth() {
return this.callTool('network_health_summary');
}
async topTalkers(minutes = 60, limit = 10) {
return this.callTool('top_talkers', { minutes, limit });
}
async searchFlows(options = {}) {
return this.callTool('search_flows', {
src_ip: options.srcIp,
dst_ip: options.dstIp,
dst_port: options.dstPort,
protocol: options.protocol,
minutes: options.minutes || 60
});
}
async runPing(target, count = 5) {
return this.callTool('run_ping_test', { target, count });
}
}
// Usage Example
async function main() {
const client = new ChompyClient(
'https://chompy.example.com',
'chmp_your_api_key_here'
);
// Get network health
const health = await client.networkHealth();
console.log('Network Health:', health.data);
// Get all Cisco devices that are up
const devices = await client.getDevices({ status: 'up', vendor: 'cisco' });
console.log(`Found ${devices.data.count} Cisco devices`);
// Run a ping test
const ping = await client.runPing('8.8.8.8');
console.log(`Ping result: ${ping.data.avg_latency_ms}ms average`);
}
main().catch(console.error);
LangChain Integration (Python)
from langchain.tools import BaseTool
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Optional, Type
import requests
class ChompyToolInput(BaseModel):
"""Input for Chompy tools"""
query: str = Field(description="Natural language query about network status")
class ChompyNetworkTool(BaseTool):
"""LangChain tool for querying Chompy network monitoring"""
name: str = "chompy_network"
description: str = """
Query network monitoring data from Chompy. Use this tool when you need to:
- Check device status or health
- View active alerts
- Analyze network traffic and top talkers
- Get interface utilization
- Run network tests (ping, traceroute, HTTP)
Input should describe what network information you need.
"""
args_schema: Type[BaseModel] = ChompyToolInput
base_url: str
api_key: str
def __init__(self, base_url: str, api_key: str):
super().__init__(base_url=base_url, api_key=api_key)
def _call_api(self, tool_name: str, params: dict = None):
response = requests.post(
f"{self.base_url}/api/mcp/tools/{tool_name}",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=params or {}
)
return response.json()
def _run(self, query: str) -> str:
query_lower = query.lower()
# Route based on query intent
if any(w in query_lower for w in ['health', 'status', 'overview', 'summary']):
result = self._call_api('network_health_summary')
return f"Network Health Summary:\n{result['data']}"
elif any(w in query_lower for w in ['alert', 'critical', 'warning']):
severity = 'critical' if 'critical' in query_lower else 'all'
result = self._call_api('get_alerts', {'severity': severity, 'status': 'active'})
return f"Active Alerts:\n{result['data']}"
elif any(w in query_lower for w in ['device', 'router', 'switch']):
result = self._call_api('get_devices', {'status': 'all', 'limit': 20})
return f"Devices:\n{result['data']}"
elif any(w in query_lower for w in ['traffic', 'bandwidth', 'top talker']):
result = self._call_api('top_talkers', {'minutes': 60, 'limit': 10})
return f"Top Talkers (last hour):\n{result['data']}"
elif any(w in query_lower for w in ['flow', 'connection']):
result = self._call_api('flow_summary', {'minutes': 60})
return f"Flow Summary:\n{result['data']}"
else:
# Default to health summary
result = self._call_api('network_health_summary')
return f"Network Health:\n{result['data']}"
# Usage with LangChain Agent
def create_network_agent():
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [
ChompyNetworkTool(
base_url="https://chompy.example.com",
api_key="chmp_your_api_key"
)
]
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
return agent
# Example usage
agent = create_network_agent()
response = agent.run("What's the current network health status?")
print(response)
OpenAI Function Calling
import openai
import json
import requests
CHOMPY_URL = "https://chompy.example.com"
CHOMPY_API_KEY = "chmp_your_api_key"
# Define functions for OpenAI
functions = [
{
"name": "get_network_health",
"description": "Get overall network health summary including device status, alerts, and traffic stats",
"parameters": {"type": "object", "properties": {}, "required": []}
},
{
"name": "get_devices",
"description": "List network devices with optional filters",
"parameters": {
"type": "object",
"properties": {
"status": {
"type": "string",
"enum": ["up", "down", "unknown", "all"],
"description": "Filter by device status"
},
"vendor": {
"type": "string",
"description": "Filter by vendor name"
}
}
}
},
{
"name": "get_alerts",
"description": "Get active alerts from the monitoring system",
"parameters": {
"type": "object",
"properties": {
"severity": {
"type": "string",
"enum": ["critical", "high", "medium", "low", "all"],
"description": "Filter by severity"
}
}
}
},
{
"name": "top_talkers",
"description": "Get top bandwidth consumers",
"parameters": {
"type": "object",
"properties": {
"minutes": {
"type": "integer",
"description": "Look back period in minutes"
},
"limit": {
"type": "integer",
"description": "Number of results"
}
}
}
},
{
"name": "run_ping_test",
"description": "Run an ICMP ping test to a target",
"parameters": {
"type": "object",
"properties": {
"target": {
"type": "string",
"description": "IP address or hostname to ping"
}
},
"required": ["target"]
}
}
]
def call_chompy(tool_name: str, params: dict = None):
"""Call Chompy API"""
response = requests.post(
f"{CHOMPY_URL}/api/mcp/tools/{tool_name}",
headers={
"Authorization": f"Bearer {CHOMPY_API_KEY}",
"Content-Type": "application/json"
},
json=params or {}
)
return response.json()
def handle_function_call(function_name: str, arguments: dict):
"""Execute the function called by OpenAI"""
tool_map = {
"get_network_health": "network_health_summary",
"get_devices": "get_devices",
"get_alerts": "get_alerts",
"top_talkers": "top_talkers",
"run_ping_test": "run_ping_test"
}
chompy_tool = tool_map.get(function_name)
if chompy_tool:
return call_chompy(chompy_tool, arguments)
return {"error": "Unknown function"}
def chat_with_network_context(user_message: str):
"""Chat with OpenAI using Chompy as a tool"""
messages = [
{"role": "system", "content": "You are a network operations assistant. Use the available functions to query network monitoring data and help users understand their network status."},
{"role": "user", "content": user_message}
]
response = openai.chat.completions.create(
model="gpt-4",
messages=messages,
functions=functions,
function_call="auto"
)
message = response.choices[0].message
# If a function was called, execute it
if message.function_call:
function_name = message.function_call.name
arguments = json.loads(message.function_call.arguments)
# Call Chompy
result = handle_function_call(function_name, arguments)
# Send result back to OpenAI
messages.append(message)
messages.append({
"role": "function",
"name": function_name,
"content": json.dumps(result)
})
final_response = openai.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
return message.content
# Example
response = chat_with_network_context("Are there any critical alerts right now?")
print(response)
Error Handling
HTTP Status Codes
| Code | Meaning |
|---|---|
200 | Success |
400 | Bad request - invalid parameters |
401 | Unauthorized - invalid or missing API key |
403 | Forbidden - insufficient permissions |
404 | Tool not found |
429 | Rate limit exceeded |
500 | Server error |
Error Response Format
{
"error": "Description of what went wrong"
}
Example Error Handling (Python)
import requests
def safe_call_tool(client, tool_name, params=None):
try:
result = client.call_tool(tool_name, params)
if result.get('success'):
return result['data']
else:
print(f"API error: {result.get('error')}")
return None
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
print("Invalid API key")
elif e.response.status_code == 403:
print("Insufficient permissions for this tool")
elif e.response.status_code == 429:
print("Rate limit exceeded - wait before retrying")
else:
print(f"HTTP error: {e}")
return None
except requests.exceptions.RequestException as e:
print(f"Connection error: {e}")
return None
Rate Limits
- Default: 1,000 requests per hour per API key
- Rate limits reset every hour from first request
- When exceeded, API returns
429 Too Many Requests
Rate Limit Headers
X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 950
X-RateLimit-Reset: 2025-12-13T11:00:00Z
Best Practices for Rate Limits
- Cache responses when possible
- Use webhooks for real-time alerts instead of polling
- Batch queries using time ranges
- Request a higher limit if needed for production use
Best Practices
1. Use Appropriate Permission Level
Request only the permission level you need:
- Use
read_onlyfor dashboards and monitoring - Use
read_writeonly if you need to run tests or acknowledge alerts - Avoid
adminunless you specifically need raw query access
2. Handle Errors Gracefully
Always implement proper error handling and retries with exponential backoff.
3. Cache When Possible
Network health summaries and device lists don't change frequently. Cache for 30-60 seconds.
4. Use Time Ranges Wisely
When querying flows or metrics:
- Use shorter time ranges (5-15 min) for real-time dashboards
- Use longer ranges (1-24 hours) for analysis and reports
5. Secure Your API Keys
- Never commit API keys to version control
- Use environment variables or secrets management
- Rotate keys periodically
- Use separate keys for different applications
6. Monitor Your Usage
Check the audit log regularly to track API usage and identify any issues.