"""
Cisco Catalyst Center (DNAC) MCP Server
This MCP server exposes Catalyst Center wireless client management tools
for use with Claude and other MCP clients.
"""
import asyncio
import logging
from typing import Any
from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
import mcp.types as types
from .wireless_client_agent import query_wireless_clients, get_client_health
from .config import load_config
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Create server instance
server = Server("dnac-wireless-clients")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
List available DNAC wireless client tools.
Returns:
List of available tools with their schemas
"""
return [
types.Tool(
name="query_wireless_clients",
description="""Query wireless clients from Cisco Catalyst Center (DNA Center) with smart limiting and guidance.
🎯 USE THIS TOOL WHEN:
- User asks about wireless clients, devices, or endpoints on the network
- Need to find specific clients by MAC, IP, hostname, or location
- Troubleshooting wireless connectivity or client issues
- Analyzing wireless network usage or client distribution
- Need to see which clients are connected to specific APs or sites
⚡ KEY FEATURES:
- Automatic result limiting (max 100 by default, configurable)
- Pre-checks total count before fetching
- Handles pagination automatically
- Provides actionable guidance when results exceed limit
- Rate limiting to prevent API throttling
📋 REQUIRED PARAMETERS:
- base_url: Catalyst Center URL (e.g., "https://dnac.example.com")
- username: DNAC username
- password: DNAC password
🔍 OPTIONAL FILTERS (use to reduce results):
- site_id: Site UUID (get from sites API)
- mac_address: Client MAC (format: AA:BB:CC:DD:EE:FF)
- hostname: Access Point hostname
- ip_address: Client IP address
- ssid: Wireless network SSID
- band: Frequency band (e.g., "2.4GHz", "5GHz")
- max_results: Maximum clients to return (default 100)
📊 RETURNS:
- clients: List of client dictionaries with full details
- total_count: Total matching clients in DNAC
- exceeded_limit: Boolean if total > max_results
- guidance: Actionable tips to refine query if limit exceeded
- fetched_count: Actual number of clients returned
💡 EXAMPLES:
Example 1 - All wireless clients (will limit if >100):
query_wireless_clients(
base_url="https://dnac.company.com",
username="admin",
password="******"
)
Example 2 - Clients at specific site:
query_wireless_clients(
base_url="https://dnac.company.com",
username="admin",
password="******",
site_id="abc-123-def-456"
)
Example 3 - Single client lookup:
query_wireless_clients(
base_url="https://dnac.company.com",
username="admin",
password="******",
mac_address="AA:BB:CC:DD:EE:FF"
)
Example 4 - Clients on specific AP:
query_wireless_clients(
base_url="https://dnac.company.com",
username="admin",
password="******",
hostname="AP-Floor2-East"
)
⚠️ IMPORTANT NOTES:
- Credentials are required for each call (consider using environment variables)
- Results are capped at max_results to prevent overload
- If total_count > max_results, guidance will suggest filter refinements
- Use get_client_health() for detailed health info on specific clients
- Always wireless clients only (family="CLIENT_TYPE_WIRELESS" auto-applied)
🔒 SECURITY:
- Use environment variables or secrets manager for credentials
- Set verify=True for production (SSL validation)
- Set verify=False only for testing with self-signed certs""",
inputSchema={
"type": "object",
"properties": {
"base_url": {
"type": "string",
"description": "Catalyst Center base URL (e.g., 'https://dnac.example.com'). Do not include /api/ path."
},
"username": {
"type": "string",
"description": "DNAC username with API access permissions"
},
"password": {
"type": "string",
"description": "DNAC password (consider using environment variables or secrets manager)"
},
"site_id": {
"type": "string",
"description": "Optional: Filter by site UUID. Use to limit results to specific location. Get site IDs from sites API."
},
"mac_address": {
"type": "string",
"description": "Optional: Filter by client MAC address (format: AA:BB:CC:DD:EE:FF). Use for single client lookup."
},
"hostname": {
"type": "string",
"description": "Optional: Filter by Access Point hostname. Shows clients on specific AP."
},
"ip_address": {
"type": "string",
"description": "Optional: Filter by client IP address (e.g., '192.168.1.100')"
},
"ssid": {
"type": "string",
"description": "Optional: Filter by SSID name. Shows clients on specific wireless network."
},
"band": {
"type": "string",
"description": "Optional: Filter by frequency band (e.g., '2.4GHz', '5GHz', '6GHz')"
},
"max_results": {
"type": "integer",
"description": "Optional: Maximum number of clients to return (default 100). Prevents overload with large result sets.",
"default": 100
},
"version": {
"type": "string",
"description": "Optional: DNAC API version (default '2.3.7.6'). Update based on your DNAC version.",
"default": "2.3.7.6"
},
"verify": {
"type": "boolean",
"description": "Optional: Verify SSL certificates (default true). Set false only for testing with self-signed certs.",
"default": True
},
"debug": {
"type": "boolean",
"description": "Optional: Enable debug logging (default true).",
"default": True
}
},
"required": ["base_url", "username", "password"]
}
),
types.Tool(
name="get_client_health",
description="""Get detailed health information for a specific wireless client from Catalyst Center.
🎯 USE THIS TOOL WHEN:
- Need detailed health metrics for a specific client
- Troubleshooting client connectivity issues
- Investigating performance problems for a known device
- Getting comprehensive client information beyond basic queries
📋 REQUIRED PARAMETERS:
- base_url: Catalyst Center URL (e.g., "https://dnac.example.com")
- username: DNAC username
- password: DNAC password
- mac_address: Client MAC address (format: AA:BB:CC:DD:EE:FF)
📊 RETURNS:
- health: Detailed health metrics and status
- response: Full API response data
- mac_address: Queried MAC address
- error: Error message if request fails
💡 EXAMPLE:
get_client_health(
base_url="https://dnac.company.com",
username="admin",
password="******",
mac_address="AA:BB:CC:DD:EE:FF"
)
⚠️ IMPORTANT NOTES:
- Requires exact MAC address of the client
- Use query_wireless_clients() first to find client MAC addresses
- Returns detailed metrics including RSSI, SNR, data rates, etc.
- Credentials required for each call
🔒 SECURITY:
- Use environment variables or secrets manager for credentials
- Set verify=True for production (SSL validation)""",
inputSchema={
"type": "object",
"properties": {
"base_url": {
"type": "string",
"description": "Catalyst Center base URL (e.g., 'https://dnac.example.com'). Do not include /api/ path."
},
"username": {
"type": "string",
"description": "DNAC username with API access permissions"
},
"password": {
"type": "string",
"description": "DNAC password (consider using environment variables or secrets manager)"
},
"mac_address": {
"type": "string",
"description": "Client MAC address (format: AA:BB:CC:DD:EE:FF)"
},
"version": {
"type": "string",
"description": "Optional: DNAC API version (default '2.3.7.6'). Update based on your DNAC version.",
"default": "2.3.7.6"
},
"verify": {
"type": "boolean",
"description": "Optional: Verify SSL certificates (default true). Set false only for testing with self-signed certs.",
"default": True
}
},
"required": ["base_url", "username", "password", "mac_address"]
}
)
]
@server.call_tool()
async def handle_call_tool(
name: str,
arguments: dict[str, Any] | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
Handle tool execution requests.
Args:
name: Name of the tool to execute
arguments: Tool arguments
Returns:
List of content items with tool results
Raises:
ValueError: If tool name is unknown
"""
if arguments is None:
arguments = {}
logger.info(f"Executing tool: {name}")
logger.debug(f"Arguments: {arguments}")
try:
if name == "query_wireless_clients":
# Execute wireless clients query
result = query_wireless_clients(**arguments)
# Format result as text content
if "error" in result and result["error"]:
content = f"❌ Error querying wireless clients:\n{result['error']}"
else:
content = format_client_query_result(result)
return [types.TextContent(
type="text",
text=content
)]
elif name == "get_client_health":
# Execute client health query
result = get_client_health(**arguments)
# Format result as text content
if "error" in result and result["error"]:
content = f"❌ Error getting client health:\n{result['error']}"
else:
content = format_client_health_result(result)
return [types.TextContent(
type="text",
text=content
)]
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Error executing tool {name}: {str(e)}", exc_info=True)
return [types.TextContent(
type="text",
text=f"❌ Error executing {name}:\n{str(e)}"
)]
def format_client_query_result(result: dict[str, Any]) -> str:
"""
Format wireless client query results for display.
Args:
result: Query result dictionary
Returns:
Formatted text output
"""
lines = []
# Header
lines.append("=" * 80)
lines.append("📡 Wireless Clients Query Results")
lines.append("=" * 80)
lines.append("")
# Summary
lines.append(f"📊 Summary:")
lines.append(f" • Total clients matching filters: {result['total_count']}")
lines.append(f" • Clients returned: {result['fetched_count']}")
lines.append(f" • Limit exceeded: {'Yes' if result['exceeded_limit'] else 'No'}")
lines.append("")
# Guidance if exceeded
if result['exceeded_limit'] and result['guidance']:
lines.append(result['guidance'])
lines.append("")
# Client details
if result['clients']:
lines.append(f"👥 Client Details ({len(result['clients'])} shown):")
lines.append("")
for i, client in enumerate(result['clients'], 1):
lines.append(f"Client {i}:")
lines.append(f" • MAC Address: {client.get('macAddress', 'N/A')}")
lines.append(f" • Hostname: {client.get('hostName', 'N/A')}")
lines.append(f" • IP Address: {client.get('hostIpV4', 'N/A')}")
lines.append(f" • SSID: {client.get('ssid', 'N/A')}")
lines.append(f" • Connected AP: {client.get('connectedNetworkDeviceName', 'N/A')}")
lines.append(f" • Band: {client.get('band', 'N/A')}")
lines.append(f" • Health Score: {client.get('healthScore', 'N/A')}")
lines.append(f" • Connection Status: {client.get('connectionStatus', 'N/A')}")
lines.append("")
else:
lines.append("ℹ️ No clients found matching the specified filters.")
lines.append("")
lines.append("=" * 80)
return "\n".join(lines)
def format_client_health_result(result: dict[str, Any]) -> str:
"""
Format client health results for display.
Args:
result: Health result dictionary
Returns:
Formatted text output
"""
lines = []
# Header
lines.append("=" * 80)
lines.append("🏥 Client Health Details")
lines.append("=" * 80)
lines.append("")
lines.append(f"📌 MAC Address: {result['mac_address']}")
lines.append("")
# Health details
health = result.get('health', {})
if health:
lines.append("📊 Health Metrics:")
for key, value in health.items():
lines.append(f" • {key}: {value}")
lines.append("")
# Full response
response = result.get('response', {})
if response:
lines.append("📋 Additional Information:")
for key, value in response.items():
if key != 'detail': # Avoid duplicating health details
lines.append(f" • {key}: {value}")
lines.append("")
if not health and not response:
lines.append("ℹ️ No health data available for this client.")
lines.append("")
lines.append("=" * 80)
return "\n".join(lines)
async def main():
"""Run the MCP server using stdio transport."""
logger.info("Starting DNAC Wireless Clients MCP Server")
# Load configuration (if available)
try:
config = load_config()
logger.info(f"Configuration loaded: {config}")
except Exception as e:
logger.warning(f"Could not load configuration: {e}")
logger.info("Server will require credentials in each request")
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
logger.info("Server initialized with stdio transport")
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="dnac-wireless-clients",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
)
)
if __name__ == "__main__":
asyncio.run(main())
