diff --git a/src/mcp_cucm_axl/route_plan.py b/src/mcp_cucm_axl/route_plan.py index 02c2bdf..edfb1a1 100644 --- a/src/mcp_cucm_axl/route_plan.py +++ b/src/mcp_cucm_axl/route_plan.py @@ -520,18 +520,32 @@ _CSS_REFERENCE_QUERIES: dict[str, dict] = { } -def find_devices_using_css(client: "AxlClient", css_name: str) -> dict: +def find_devices_using_css( + client: "AxlClient", + css_name: str, + max_per_category: int = 50, +) -> dict: """Impact analysis: enumerate every reference to this CSS across the schema. Useful before changing or removing a CSS — answers "what would break if I touched this CSS?" Cross-references the major schema tables that hold fkcallingsearchspace* columns. - Returns groups of references by category. Each reference has: - - name: the object identifier (pattern text, device name, or DN) - - context: partition (for patterns) or device class (for devices) - - description: free-text description - - _table, _column: which fk-column referenced this CSS + Args: + css_name: The exact CSS name (case-sensitive). + max_per_category: Max rows to return per reference category. Default 50. + On heavily-referenced CSSs (e.g., Internal-CSS with hundreds of + line-CFA references), the un-paginated response can exceed MCP + response limits. Each truncated category includes a `truncated: + true` flag so the LLM knows to drill in (call again with higher + max, or use `axl_sql` directly for that specific table+column). + + Returns: + references_by_category: each entry has: + - rows: list of references, capped at max_per_category + - returned_count: number of rows in this response + - truncated: True if the underlying result has more rows than returned + - table, column: which fk-column was queried """ safe = _esc(css_name) css_lookup = client.execute_sql_query( @@ -542,29 +556,51 @@ def find_devices_using_css(client: "AxlClient", css_name: str) -> dict: css_pkid = css_lookup["rows"][0]["pkid"] safe_pkid = _esc(css_pkid) - grouped: dict[str, list[dict]] = {} - for label, spec in _CSS_REFERENCE_QUERIES.items(): - sql = spec["sql"].format(pkid=safe_pkid) - try: - result = client.execute_sql_query(sql) - if result["rows"]: - grouped[label] = [ - {**r, "_table": spec["table"], "_column": spec["column"]} - for r in result["rows"] - ] - except Exception as e: - # Don't let one failed reference point block the whole audit - grouped[label] = [{ - "_error": str(e)[:200], - "_table": spec["table"], - "_column": spec["column"], - }] + # Trick: ask Informix for max+1 rows. If we get max+1 back, we know the + # underlying result has MORE than max — set truncated=True and trim. + # This avoids the cost of a separate COUNT(*) per category. + fetch_n = max_per_category + 1 - total = sum(len(v) for v in grouped.values()) + grouped: dict[str, dict] = {} + for label, spec in _CSS_REFERENCE_QUERIES.items(): + # Inject a FIRST limit into the SELECT — the queries all start with + # `SELECT ` so a literal replacement is reliable here. + base_sql = spec["sql"].format(pkid=safe_pkid).strip() + if base_sql.upper().startswith("SELECT "): + limited_sql = "SELECT FIRST " + str(fetch_n) + " " + base_sql[7:] + else: + limited_sql = base_sql + + try: + result = client.execute_sql_query(limited_sql) + rows = result["rows"] + if not rows: + continue + truncated = len(rows) > max_per_category + if truncated: + rows = rows[:max_per_category] + grouped[label] = { + "table": spec["table"], + "column": spec["column"], + "returned_count": len(rows), + "truncated": truncated, + "rows": rows, + } + except Exception as e: + grouped[label] = { + "table": spec["table"], + "column": spec["column"], + "error": str(e)[:200], + } + + total_returned = sum(c.get("returned_count", 0) for c in grouped.values()) + any_truncated = any(c.get("truncated") for c in grouped.values()) return { "css_name": css_name, "css_pkid": css_pkid, - "total_references": total, + "total_returned": total_returned, + "any_truncated": any_truncated, + "max_per_category": max_per_category, "references_by_category": grouped, } @@ -573,21 +609,38 @@ def find_devices_using_css(client: "AxlClient", css_name: str) -> dict: # Route Filters # ==================================================================== -def list_route_filters(client: "AxlClient", name: str | None = None) -> dict: - """List route filters with their member clauses. +def list_route_filters( + client: "AxlClient", + name: str | None = None, + include_members: bool = False, +) -> dict: + """List route filters with their composition rules. Route filters compose with @-pattern (NANPA) route patterns to constrain matches — e.g., a filter that matches only when AREA-CODE == 208 narrows - a `9.@` pattern to in-state calls. Members specify (digit, operator, tag) - triples; the clause column shows the assembled expression. + a `9.@` pattern to in-state calls. The `clause` field contains the + pre-rendered expression, which is usually enough for audit work. + + Args: + name: If given, return only the named filter. + include_members: If True, also fetch the (digit, operator, tag) member + rules for each filter. Default False — the response can be very + large (hundreds of KB on clusters with many international filters) + and the `clause` field already shows the human-readable rule. Set + True only when you need to programmatically reconstruct or modify + the filter logic. """ where = f"WHERE rf.name = '{_esc(name)}'" if name else "" + # Member count is cheap to compute server-side and gives auditors a quick + # signal for which filters are complex without paying for the full member + # listing. sql = f""" SELECT rf.name AS filter_name, rf.clause AS clause, dp.name AS dial_plan, - rf.pkid AS pkid + rf.pkid AS pkid, + (SELECT COUNT(*) FROM routefiltermember rfm WHERE rfm.fkroutefilter = rf.pkid) AS member_count FROM routefilter rf LEFT OUTER JOIN dialplan dp ON rf.fkdialplan = dp.pkid {where} @@ -597,28 +650,30 @@ def list_route_filters(client: "AxlClient", name: str | None = None) -> dict: if not result["rows"]: return {"filter_count": 0, "route_filters": []} - # Fetch members per filter out = [] for filt in result["rows"]: - member_sql = f""" - SELECT - dpt.tag AS tag, - op.name AS operator, - rfm.digits AS digits, - rfm.precedence AS precedence - FROM routefiltermember rfm - LEFT OUTER JOIN dialplantag dpt ON rfm.fkdialplantag = dpt.pkid - LEFT OUTER JOIN typeoperator op ON rfm.tkoperator = op.enum - WHERE rfm.fkroutefilter = '{_esc(filt["pkid"])}' - ORDER BY rfm.precedence - """ - members = client.execute_sql_query(member_sql) - out.append({ + entry = { "name": filt.get("filter_name"), "clause": filt.get("clause"), "dial_plan": filt.get("dial_plan"), - "members": members["rows"], - }) + "member_count": _to_int(filt.get("member_count")), + } + if include_members: + member_sql = f""" + SELECT + dpt.tag AS tag, + op.name AS operator, + rfm.digits AS digits, + rfm.precedence AS precedence + FROM routefiltermember rfm + LEFT OUTER JOIN dialplantag dpt ON rfm.fkdialplantag = dpt.pkid + LEFT OUTER JOIN typeoperator op ON rfm.tkoperator = op.enum + WHERE rfm.fkroutefilter = '{_esc(filt["pkid"])}' + ORDER BY rfm.precedence + """ + members = client.execute_sql_query(member_sql) + entry["members"] = members["rows"] + out.append(entry) return {"filter_count": len(out), "route_filters": out} diff --git a/src/mcp_cucm_axl/server.py b/src/mcp_cucm_axl/server.py index a5a1947..67bddf2 100644 --- a/src/mcp_cucm_axl/server.py +++ b/src/mcp_cucm_axl/server.py @@ -263,7 +263,7 @@ def route_device_pool_route_groups(device_pool_name: str | None = None) -> dict: @mcp.tool -def route_devices_using_css(css_name: str) -> dict: +def route_devices_using_css(css_name: str, max_per_category: int = 50) -> dict: """Impact analysis: every reference to a CSS across the schema. Use before changing or removing a CSS to find all dependent devices, lines, @@ -272,23 +272,29 @@ def route_devices_using_css(css_name: str) -> dict: Args: css_name: The exact CSS name (case-sensitive). + max_per_category: Max rows per category. Default 50. Each category + with more than `max_per_category` references gets `truncated: true` + so you know to either raise the limit or drill in via `axl_sql`. """ - return route_plan.find_devices_using_css(_client(), css_name) + return route_plan.find_devices_using_css(_client(), css_name, max_per_category) @mcp.tool -def route_filters(name: str | None = None) -> dict: +def route_filters(name: str | None = None, include_members: bool = False) -> dict: """List route filters with their composition rules. Route filters compose with @-pattern (NANPA) route patterns to constrain which calls match — e.g., "AREA-CODE == 208" narrows a `9.@` pattern to - in-state calls. Each filter has an ordered list of (digit, operator, tag) - member clauses. + in-state calls. Args: name: Optional. If given, return only the named filter. + include_members: If True, include the (digit, operator, tag) member + rules. Default False — the `clause` field already shows the + human-readable rule, and member listings can be very large + (hundreds of KB across all filters on a complex cluster). """ - return route_plan.list_route_filters(_client(), name) + return route_plan.list_route_filters(_client(), name, include_members) # ====================================================================