mcaxl/src/mcp_cucm_axl/normalize.py
Ryan Malloy 8b3da9d729 Initial mcp-cucm-axl
Read-only MCP server for Cisco Unified CM 15 AXL — built for LLM-driven
cluster auditing, with a particular focus on the Route Plan Report:
partitions, calling search spaces, route patterns, translation patterns,
called/calling party transformations, and digit-discard instructions.

Pairs intentionally with the sibling mcp-cisco-docs server (live
cluster state + vendor docs in one LLM context).

Architecture:
  - zeep SOAP client to CUCM AXL
  - WSDL bootstrap from Cisco's axlsqltoolkit.zip (auto-extract on
    first launch; zip is gitignored, vendor-licensed)
  - SQLite response cache at ~/.cache/mcp-cucm-axl/responses/
  - Schema-grounded prompts that pull chunks from the sibling
    cisco-docs index (docs_loader.py)

Read-only by structural guarantee — never registers AXL write methods
(no executeSQLUpdate, no add*/update*/remove*/apply*/reset*/restart*
tools). SQL queries also client-side validated (sql_validator.py) to
begin with SELECT or WITH.

Tools exposed:
  Foundational: axl_version, axl_sql, axl_list_tables,
                axl_describe_table, cache_stats, cache_clear
  Route plan:   route_partitions, route_calling_search_spaces,
                route_patterns, route_inspect_pattern,
                route_lists_and_groups, route_translation_chain,
                route_digit_discard_instructions

Prompts (schema-grounded):
  route_plan_overview, investigate_pattern, audit_routing,
  cucm_sql_help

Tests cover cache, docs_loader, normalize, sql_validator, wildcard.
2026-04-25 20:29:18 -06:00

126 lines
4.2 KiB
Python

"""Normalize Informix-flavored result values for LLM consumption.
CUCM's Informix database returns several encodings that are awkward for
an LLM to interpret correctly without help:
- Booleans as 't'/'f' strings (not native booleans).
- Foreign-key codes like `tkreleasecausevalue='0'` that need a separate
join to a `type*` table to get a human-readable name.
This module:
1. Converts 't'/'f' to True/False for known boolean columns.
2. Provides `TypeDecoder` — a small lazy cache that looks up tk* enum
codes against their CUCM `type*` lookup tables on first use, so the
output gets the human-readable name without us hardcoding any mappings.
The decoder reads from CUCM's own `typereleasecausevalue` etc. tables, so
it stays correct across CUCM versions without maintenance.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .client import AxlClient
# Columns we know to be boolean from the CUCM data dictionary. Adding a
# column here is safe — if it's not actually a 't'/'f' boolean, normalization
# is a no-op (only literal 't' and 'f' strings get coerced).
BOOLEAN_COLUMNS: frozenset[str] = frozenset({
"block_enabled",
"blockenable",
"isemergencyservicenumber",
"isanonymous",
"iscallable",
"ismessagewaitingon",
"supportoverlapsending",
"outsidedialtone",
"rejectanonymouscall",
"routethisdialplan",
"usecallercss",
"useoriginatorcss",
"withtag",
"withvalueclause",
"personalroutingenabled",
})
def normalize_bool(value: object) -> object:
"""'t'/'f' → True/False; passthrough everything else."""
if value == "t":
return True
if value == "f":
return False
return value
def normalize_row(row: dict) -> dict:
"""Apply boolean normalization to known boolean columns in a row dict."""
out = {}
for k, v in row.items():
out[k] = normalize_bool(v) if k in BOOLEAN_COLUMNS else v
return out
def normalize_rows(rows: list[dict]) -> list[dict]:
return [normalize_row(r) for r in rows]
class TypeDecoder:
"""Lazily resolve `tk*` foreign-key codes to their human-readable names.
Usage:
decoder = TypeDecoder(client)
name = decoder.decode("tkreleasecausevalue", "0") # → "No Error"
On first call for a given tk* code, queries the matching `type*` table
and caches the {enum: name} mapping. Subsequent calls hit the cache.
"""
# Mapping from tk* column → type* lookup table. Add as we encounter
# new codes that benefit from decoding. Codes not in this map remain
# as the raw integer string (still queryable, just not decoded).
TK_TO_TYPE: dict[str, str] = {
"tkpatternusage": "typepatternusage",
"tkreleasecausevalue": "typereleasecausevalue",
"tkpatternprecedence": "typepatternprecedence",
"tkpatternrouteclass": "typepatternrouteclass",
"tkclass": "typeclass",
"tkdeviceprotocol": "typedeviceprotocol",
"tkmodel": "typemodel",
"tkproduct": "typeproduct",
"tkstatus_partyentrancetone": "typestatus_partyentrancetone",
}
def __init__(self, client: "AxlClient"):
self.client = client
self._cache: dict[str, dict[str, str]] = {}
def _load_table(self, type_table: str) -> dict[str, str]:
if type_table in self._cache:
return self._cache[type_table]
try:
result = self.client.execute_sql_query(
f"SELECT enum, name FROM {type_table}"
)
mapping = {
str(r.get("enum")): r.get("name", "")
for r in result.get("rows", [])
if r.get("enum") is not None
}
except Exception:
mapping = {}
self._cache[type_table] = mapping
return mapping
def decode(self, tk_column: str, code: object) -> object:
"""Return the human-readable name for code, or the original code if unknown."""
if code is None or code == "":
return code
type_table = self.TK_TO_TYPE.get(tk_column)
if type_table is None:
return code
mapping = self._load_table(type_table)
return mapping.get(str(code), code)