"""
Catalyst Center (DNA Center) Wireless Client Tool
This module provides intelligent tools for querying Cisco Catalyst Center (DNAC)
wireless clients with automatic limiting, pagination, and actionable guidance.
"""
from dnacentersdk import api
from dnacentersdk.exceptions import ApiError
from typing import Dict, List, Any, Optional, Union
import time
import logging
import json
from datetime import datetime
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler('catalyst_center_debug.log')
]
)
logger = logging.getLogger(__name__)
class WirelessClientAgent:
"""
Agent for querying DNAC wireless clients with auto-limiting and guidance.
Features:
- Enforces hard limit on results (default 100)
- Pre-checks total count before fetching
- Handles pagination automatically
- Provides actionable guidance when limits exceeded
- Rate limiting to prevent API throttling
Usage:
agent = WirelessClientAgent(dnac_client, max_results=100)
result = agent.get_clients(site_id="...", max_results=100)
"""
def __init__(
self,
dnac_client: api.DNACenterAPI,
max_results: int = 100,
rate_limit_delay: float = 0.5,
debug: bool = True
):
"""
Initialize the WirelessClientAgent.
Args:
dnac_client: Initialized DNACenterAPI instance
max_results: Hard cap on returned clients (default 100)
rate_limit_delay: Seconds to sleep between API calls (DNAC throttling)
debug: Enable debug logging (default True)
"""
self.dnac = dnac_client
self.max_results = max_results
self.rate_limit_delay = rate_limit_delay
self.debug = debug
if self.debug:
logger.info(f"WirelessClientAgent initialized: max_results={max_results}, rate_limit_delay={rate_limit_delay}")
logger.debug(f"DNAC client base_url: {dnac_client.base_url if hasattr(dnac_client, 'base_url') else 'unknown'}")
def _get_count(self, **filters) -> int:
"""
Get total count of clients matching filters.
Args:
**filters: Query filters for wireless clients
Returns:
Total count of matching clients
"""
if self.debug:
logger.debug(f"Getting client count with filters: {json.dumps(filters, indent=2)}")
time.sleep(self.rate_limit_delay)
try:
logger.debug("Calling dnac.clients.retrieves_the_total_count_of_clients_by_applying_basic_filtering()")
count_resp = self.dnac.clients.retrieves_the_total_count_of_clients_by_applying_basic_filtering(**filters)
if self.debug:
logger.debug(f"Count API response: {json.dumps(count_resp, indent=2, default=str)}")
response_data = count_resp.get('response', {})
if isinstance(response_data, dict):
count = response_data.get('count', 0)
else:
count = response_data if isinstance(response_data, int) else 0
logger.info(f"Total client count: {count}")
return count
except ApiError as e:
logger.error(f"DNAC API Error getting count: {e.status_code} - {e.body}")
logger.error(f"Full error: {str(e)}", exc_info=True)
raise Exception(f"Failed to get client count: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error getting count: {str(e)}", exc_info=True)
raise Exception(f"Failed to get client count: {str(e)}")
def _fetch_paginated(self, **filters) -> List[Dict]:
"""
Fetch up to max_results clients.
Note: This DNAC version doesn't support limit/offset parameters.
The API returns a default page (typically 100-500 clients) and we truncate to max_results.
Args:
**filters: Query filters for wireless clients
Returns:
List of client dictionaries (up to max_results)
"""
logger.info(f"Fetching clients (max_results={self.max_results})")
time.sleep(self.rate_limit_delay)
try:
logger.debug(f"Calling dnac.clients.retrieves_the_list_of_clients_while_also_offering_basic_filtering_and_sorting_capabilities(filters={filters})")
resp = self.dnac.clients.retrieves_the_list_of_clients_while_also_offering_basic_filtering_and_sorting_capabilities(
**filters
)
if self.debug:
logger.debug(f"API response keys: {list(resp.keys())}")
logger.debug(f"Full response preview: {json.dumps(resp, indent=2, default=str)[:1000]}...")
clients = resp.get('response', [])
logger.info(f"Retrieved {len(clients)} clients from API")
if not clients:
logger.warning("No clients returned from API")
return []
final_clients = clients[:self.max_results]
logger.info(f"Returning {len(final_clients)} clients (truncated to max_results={self.max_results})")
if self.debug and final_clients:
logger.debug(f"Sample client (first): {json.dumps(final_clients[0], indent=2, default=str)[:500]}...")
return final_clients
except ApiError as e:
logger.error(f"DNAC API Error fetching clients: {e.status_code} - {e.body}")
logger.error(f"Full API error: {str(e)}", exc_info=True)
if hasattr(e, 'response') and e.response:
logger.error(f"Response headers: {e.response.headers if hasattr(e.response, 'headers') else 'N/A'}")
logger.error(f"Response text: {e.response.text if hasattr(e.response, 'text') else 'N/A'}")
raise Exception(f"Failed to fetch clients: {str(e)}")
except AttributeError as e:
logger.error(f"SDK error parsing API response: {str(e)}")
logger.error(f"This usually means the API returned an error in an unexpected format")
logger.error(f"The API endpoint might not support the parameters we're using")
logger.error("Full error:", exc_info=True)
raise Exception(f"API returned error in unexpected format. Check if DNAC API version supports these parameters: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error fetching clients: {str(e)}", exc_info=True)
raise Exception(f"Failed to fetch clients: {str(e)}")
def get_clients(
self,
**filters
) -> Dict[str, Union[List[Dict], int, str, bool]]:
"""
Smart query for wireless clients with automatic limiting and guidance.
Args:
**filters: Query filters including:
- type: Client type (automatically set to "WIRELESS")
- site_id: Global/site ID (UUID)
- connected_network_device_name: Access Point/device name
- mac_address: Client MAC address
- ipv4_address: Client IPv4 address
- ipv6_address: Client IPv6 address
- ssid: Network SSID
- band: Frequency band (e.g., "2.4GHz", "5GHz")
- wlc_name: Wireless LAN Controller name
Returns:
Dictionary with:
- clients: List of client dictionaries
- total_count: Total matching clients
- exceeded_limit: Boolean if total > max_results
- guidance: String with tips if exceeded
- fetched_count: Actual number returned
- error: Error message if failed
"""
logger.info("="*80)
logger.info(f"get_clients() called at {datetime.now().isoformat()}")
logger.info(f"Filters provided: {json.dumps(filters, indent=2)}")
logger.info("="*80)
try:
logger.info("Step 1: Pre-checking total count...")
total_count = self._get_count(**filters)
exceeded = total_count > self.max_results
logger.info(f"Total count: {total_count}, Max results: {self.max_results}, Exceeded: {exceeded}")
if total_count > 0:
logger.info("Step 2: Fetching clients (count > 0)...")
clients = self._fetch_paginated(**filters)
else:
logger.warning("Total count is 0, skipping fetch")
clients = []
guidance = ""
if exceeded:
logger.info("Step 3: Generating guidance (limit exceeded)...")
guidance = self._generate_guidance(filters, total_count)
result = {
'clients': clients,
'total_count': total_count,
'exceeded_limit': exceeded,
'guidance': guidance,
'fetched_count': len(clients)
}
logger.info("="*80)
logger.info(f"get_clients() completed successfully")
logger.info(f"Result summary: total_count={total_count}, fetched_count={len(clients)}, exceeded={exceeded}")
logger.info("="*80)
if self.debug and clients:
logger.debug(f"Sample client data (first client): {json.dumps(clients[0], indent=2, default=str)}")
return result
except Exception as e:
logger.error("="*80)
logger.error(f"get_clients() FAILED with exception")
logger.error(f"Exception type: {type(e).__name__}")
logger.error(f"Exception message: {str(e)}")
logger.error("Full traceback:", exc_info=True)
logger.error("="*80)
return {
'error': str(e),
'clients': [],
'total_count': 0,
'exceeded_limit': False,
'guidance': '',
'fetched_count': 0
}
def _generate_guidance(self, filters: Dict, total_count: int) -> str:
"""
Generate dynamic guidance to help reduce result count.
Args:
filters: Current query filters
total_count: Total matching clients
Returns:
Multi-line guidance string with actionable tips
"""
tips = [
f"⚠️ Total clients: {total_count}. Use more specific filters to get <{self.max_results}:",
"",
"🎯 Recommended filters to add:",
" • site_id='<site-uuid>' - Limit to specific site/building",
" • mac_address='AA:BB:CC:DD:EE:FF' - Query single client",
" • hostname='<AP-Name>' - Filter by Access Point",
" • ip_address='192.168.1.100' - Filter by client IP",
" • ssid='<network-name>' - Filter by wireless network",
" • band='5GHz' - Filter by frequency band",
"",
"📊 Additional resources:",
" • Get sites: dnac.sites.get_site()",
" • Client health: dnac.clients.get_overall_client_health(mac_address='...')",
" • Client detail: dnac.clients.get_client_detail(mac_address='...')"
]
current_filters = [f"{k}={repr(v)}" for k, v in filters.items() if v]
if current_filters:
tips.insert(2, f"📌 Current filters: {', '.join(current_filters)}")
tips.insert(3, "")
return "\n".join(tips)
def query_wireless_clients(
base_url: str,
username: str,
password: str,
site_id: Optional[str] = None,
mac_address: Optional[str] = None,
hostname: Optional[str] = None,
ip_address: Optional[str] = None,
ssid: Optional[str] = None,
band: Optional[str] = None,
max_results: int = 100,
version: str = "2.3.7.6",
verify: bool = True,
debug: bool = True
) -> Dict[str, Any]:
"""
Query wireless clients from Cisco Catalyst Center (DNA Center).
This function automatically handles:
- Connection to DNAC with credentials
- Pagination for large result sets
- Result limiting (default 100 max)
- Actionable guidance when results exceed limit
Args:
base_url: DNAC URL (e.g., "https://dnac.example.com")
username: DNAC username
password: DNAC password
site_id: Filter by site UUID (optional)
mac_address: Filter by client MAC address (optional)
hostname: Filter by AP hostname (optional)
ip_address: Filter by client IP (optional)
ssid: Filter by SSID name (optional)
band: Filter by frequency band (optional)
max_results: Maximum clients to return (default 100)
version: DNAC API version (default "2.3.7.6")
verify: Verify SSL certificates (default True)
debug: Enable debug logging (default True)
Returns:
Dictionary with client list, metadata, and guidance
Raises:
Exception: If DNAC connection or query fails
"""
logger.info("="*80)
logger.info(f"query_wireless_clients() called at {datetime.now().isoformat()}")
logger.info(f"Connection: base_url={base_url}, username={username}, version={version}, verify={verify}")
logger.info("="*80)
try:
logger.info("Initializing DNAC client...")
logger.debug(f"DNAC connection params: base_url={base_url}, version={version}, verify={verify}")
dnac = api.DNACenterAPI(
base_url=base_url,
username=username,
password=password,
version=version,
verify=verify
)
logger.info("✓ DNAC client initialized successfully")
logger.info(f"Creating WirelessClientAgent with max_results={max_results}, debug={debug}")
agent = WirelessClientAgent(dnac, max_results=max_results, debug=debug)
filters = {}
if site_id:
filters['site_id'] = site_id
if mac_address:
filters['mac_address'] = mac_address
if hostname:
filters['connected_network_device_name'] = hostname
if ip_address:
filters['ipv4_address'] = ip_address
if ssid:
filters['ssid'] = ssid
if band:
filters['band'] = band
logger.info(f"Built filters: {json.dumps(filters, indent=2)}")
logger.info("Calling agent.get_clients()...")
result = agent.get_clients(**filters)
logger.info(f"query_wireless_clients() completed: fetched {result.get('fetched_count', 0)} clients")
return result
except Exception as e:
logger.error("="*80)
logger.error(f"query_wireless_clients() FAILED")
logger.error(f"Exception: {type(e).__name__}: {str(e)}")
logger.error("Full traceback:", exc_info=True)
logger.error("="*80)
return {
'error': f"Failed to query wireless clients: {str(e)}",
'clients': [],
'total_count': 0,
'exceeded_limit': False,
'guidance': '',
'fetched_count': 0
}
def get_client_health(
base_url: str,
username: str,
password: str,
mac_address: str,
version: str = "2.3.7.6",
verify: bool = True
) -> Dict[str, Any]:
"""
Get detailed health information for a specific wireless client.
Args:
base_url: DNAC URL (e.g., "https://dnac.example.com")
username: DNAC username
password: DNAC password
mac_address: Client MAC address
version: DNAC API version (default "2.3.7.6")
verify: Verify SSL certificates (default True)
Returns:
Dictionary with client health details
"""
try:
dnac = api.DNACenterAPI(
base_url=base_url,
username=username,
password=password,
version=version,
verify=verify
)
response = dnac.clients.retrieves_specific_client_information_matching_the_macaddress(
id=mac_address
)
return {
'health': response.get('detail', {}),
'response': response.get('response', {}),
'mac_address': mac_address
}
except Exception as e:
return {
'error': f"Failed to get client health: {str(e)}",
'mac_address': mac_address
}
