Extract prompts into a package + add sip_trunk_report
Refactor: the four existing inline prompts in server.py move into
individual modules under src/mcp_cucm_axl/prompts/. Server.py keeps
thin @mcp.prompt-decorated shims that delegate to the corresponding
render() function — FastMCP needs the shims because it introspects
their signatures to expose parameters to the LLM, but the prompt
*content* now lives one-prompt-per-file.
Why: server.py's prompt section had grown to ~200 lines of inline
markdown. As more query patterns get documented (see
docs/query-patterns/) this would only worsen. Per-module bodies are
easier to diff, review, and unit-test in isolation.
Layout:
src/mcp_cucm_axl/prompts/
__init__.py
_common.py — shared helpers, keyword sets, render_schema_block
route_plan_overview.py
investigate_pattern.py
audit_routing.py
cucm_sql_help.py
sip_trunk_report.py — NEW
Each prompt module exports a `render(docs, *args) -> str` function
that takes the DocsIndex as a parameter (no module globals). The
shim in server.py grabs the runtime `_docs` and passes it in. Pure
functions = trivially unit-testable.
NEW prompt: sip_trunk_report.
Implementation reference: docs/query-patterns/sip-trunk-report.md
(written separately as a query-pattern doc, validated against the
live cluster). The prompt embeds:
- Step 1: trunk inventory SQL (device + sipdevice + 5 LEFT JOINs)
- Step 2: per-destination SQL (siptrunkdestination)
- Step 3: pointer to existing route_lists_and_groups() tool
- Step 4: findings template (SPOF, profile sprawl, CSS asymmetry,
codec heterogeneity, DNS-vs-IP, security posture)
Optional `name_filter` parameter narrows the inventory via LIKE; the
filter value is escaped for SQL safety (single quotes doubled per
Informix convention).
Tests: 14 new in tests/test_prompts_package.py covering each
prompt's render() with and without docs, plus a registration smoke
test that confirms the FastMCP shim set matches the prompts package
exports (catches the case where a new module is added without its
shim).
Total: 100 → 114 tests; 5 prompts registered; live verification
against cucm-pub.binghammemorial.org confirms the embedded SQL
produces real inventory data. The four original prompts are
behaviorally identical to before — same content, just relocated.
This commit is contained in:
parent
2690c2225b
commit
e6aa075793
34
src/mcp_cucm_axl/prompts/__init__.py
Normal file
34
src/mcp_cucm_axl/prompts/__init__.py
Normal file
@ -0,0 +1,34 @@
|
||||
"""Schema-grounded conversation seeds for `mcp-cucm-axl`.
|
||||
|
||||
Each module here defines a `render(docs, *args, **kwargs) -> str` function
|
||||
that produces the prompt body. The `@mcp.prompt` registration shims live
|
||||
in `server.py` — they're thin wrappers that pull the module-global `_docs`
|
||||
and delegate. Keeping the rendering pure (no module globals, no FastMCP
|
||||
imports here) makes prompts unit-testable in isolation and keeps each
|
||||
prompt's content in its own file.
|
||||
|
||||
To add a new prompt:
|
||||
1. Create `src/mcp_cucm_axl/prompts/<name>.py` exporting a `render()`.
|
||||
2. Re-export it below.
|
||||
3. Add a thin `@mcp.prompt`-decorated shim in `server.py` that calls it.
|
||||
|
||||
Adding the shim is required because FastMCP introspects the *decorated*
|
||||
function's signature to expose parameters to the LLM — the registration
|
||||
shim is where the parameter contract lives.
|
||||
"""
|
||||
|
||||
from . import (
|
||||
audit_routing,
|
||||
cucm_sql_help,
|
||||
investigate_pattern,
|
||||
route_plan_overview,
|
||||
sip_trunk_report,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"audit_routing",
|
||||
"cucm_sql_help",
|
||||
"investigate_pattern",
|
||||
"route_plan_overview",
|
||||
"sip_trunk_report",
|
||||
]
|
||||
73
src/mcp_cucm_axl/prompts/_common.py
Normal file
73
src/mcp_cucm_axl/prompts/_common.py
Normal file
@ -0,0 +1,73 @@
|
||||
"""Shared helpers and constants used by multiple prompts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..docs_loader import DocsIndex
|
||||
|
||||
|
||||
# Keyword sets for pulling relevant doc chunks. Tuned per audit topic so
|
||||
# prompt enrichment focuses on the right schema docs without burning tokens
|
||||
# on irrelevant CLI-reference material.
|
||||
ROUTE_KEYWORDS = [
|
||||
"route plan", "route pattern", "translation pattern",
|
||||
"calling search space", "partition", "transformation",
|
||||
"digit discard", "numplan", "routepartition",
|
||||
]
|
||||
|
||||
AUDIT_KEYWORDS = {
|
||||
"full": ROUTE_KEYWORDS,
|
||||
"translations": [
|
||||
"translation pattern", "called party transformation",
|
||||
"calling party transformation", "digit discard",
|
||||
],
|
||||
"css_partitions": [
|
||||
"calling search space", "partition", "css",
|
||||
],
|
||||
"transformations": [
|
||||
"called party transformation", "calling party transformation",
|
||||
"transformation mask", "prefix digits",
|
||||
],
|
||||
"route_lists": [
|
||||
"route list", "route group", "device pool", "local route group",
|
||||
],
|
||||
}
|
||||
|
||||
SIP_TRUNK_KEYWORDS = [
|
||||
"sip trunk", "sip device", "sipdevice", "siptrunkdestination",
|
||||
"sip profile", "trunk", "transport", "early offer",
|
||||
]
|
||||
|
||||
|
||||
def docs_or_empty_msg() -> str:
|
||||
"""Fallback when the docs index isn't loaded — tells the LLM how to
|
||||
get equivalent info via the sibling cisco-docs MCP server."""
|
||||
return (
|
||||
"_The cisco-docs index is not loaded. Set CISCO_DOCS_INDEX_PATH or "
|
||||
"ensure /home/rpm/bingham/docs/src/assets/.cisco-docs-index/ exists. "
|
||||
"You can also use the sibling `cisco-docs` MCP server's `search_docs` "
|
||||
"tool for live semantic search._"
|
||||
)
|
||||
|
||||
|
||||
def render_schema_block(
|
||||
docs: "DocsIndex | None",
|
||||
keywords: list[str],
|
||||
*,
|
||||
max_chunks: int = 5,
|
||||
max_chars_per_chunk: int = 1000,
|
||||
) -> str:
|
||||
"""Pull doc chunks for the given keywords and format them for embedding.
|
||||
|
||||
Returns a `docs_or_empty_msg()` notice when docs is None.
|
||||
"""
|
||||
if docs is None:
|
||||
return docs_or_empty_msg()
|
||||
chunks = docs.find(
|
||||
keywords,
|
||||
max_chunks=max_chunks,
|
||||
max_chars_per_chunk=max_chars_per_chunk,
|
||||
)
|
||||
return docs.format_chunks_for_prompt(chunks)
|
||||
76
src/mcp_cucm_axl/prompts/audit_routing.py
Normal file
76
src/mcp_cucm_axl/prompts/audit_routing.py
Normal file
@ -0,0 +1,76 @@
|
||||
"""Comprehensive routing audit walkthrough."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from ._common import AUDIT_KEYWORDS, ROUTE_KEYWORDS, render_schema_block
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..docs_loader import DocsIndex
|
||||
|
||||
|
||||
def render(docs: "DocsIndex | None", focus: str = "full") -> str:
|
||||
"""Conduct a focused audit of the cluster's routing configuration.
|
||||
|
||||
Args:
|
||||
focus: One of "full", "translations", "css_partitions",
|
||||
"transformations", "route_lists". Tunes which schema chunks
|
||||
get embedded.
|
||||
"""
|
||||
keyword_set = AUDIT_KEYWORDS.get(focus, ROUTE_KEYWORDS)
|
||||
schema_block = render_schema_block(
|
||||
docs, keyword_set, max_chunks=6, max_chars_per_chunk=900
|
||||
)
|
||||
|
||||
return f"""# CUCM Routing Audit — focus: `{focus}`
|
||||
|
||||
Conduct a focused audit of the cluster's routing configuration. Goal: produce
|
||||
an actionable findings report — not just a description of the config.
|
||||
|
||||
## Audit checklist
|
||||
|
||||
### Partitions and access control
|
||||
- [ ] Are there partitions with zero patterns? (legacy/orphaned)
|
||||
- [ ] Are there partitions not referenced by any CSS? (unreachable)
|
||||
- [ ] Does the partition naming convention reflect actual scope?
|
||||
|
||||
### Calling Search Spaces
|
||||
- [ ] CSS-to-CSS comparison: any near-duplicates that differ only in order?
|
||||
- [ ] CSSs that include partitions in surprising orders (longest-match implications)
|
||||
- [ ] CSSs that are referenced by zero devices (use axl_sql against device table)
|
||||
|
||||
### Translation patterns
|
||||
- [ ] What does each translation pattern actually transform? Any with no
|
||||
transformation that exist purely for partition routing?
|
||||
- [ ] Calling-party transformations applied at translation: are they
|
||||
documented? Why is the calling number being rewritten?
|
||||
- [ ] Translation chains: do any translations route into partitions where
|
||||
another translation will match again? (chains can be intentional but
|
||||
obscure caller-ID and routing logic)
|
||||
|
||||
### Route patterns
|
||||
- [ ] Wildcard breadth: any `9.@` (NANPA at-sign) patterns? Are they intentional?
|
||||
- [ ] Block patterns: which patterns have `block_enabled` set? What are they
|
||||
blocking and why?
|
||||
- [ ] Patterns with no description — flag for documentation.
|
||||
|
||||
### Transformations (called party / calling party)
|
||||
- [ ] Digit discard instructions: which DDIs are referenced? Are any DDIs
|
||||
defined but never used?
|
||||
- [ ] Prefix-digits-out: any unusual prefixes (international, special service)?
|
||||
- [ ] Calling-party masks that hide internal extensions on outbound calls.
|
||||
|
||||
### Route lists and groups
|
||||
- [ ] Route lists with only one route group: simple, fine.
|
||||
- [ ] Route lists with many: walk the failover order, confirm it's intentional.
|
||||
- [ ] Route groups containing devices that are unregistered or disabled.
|
||||
|
||||
## Reference: CUCM data dictionary
|
||||
|
||||
{schema_block}
|
||||
|
||||
Run the relevant tool calls now and produce a structured findings report
|
||||
with category headers, observation, severity (info/warning/error), and
|
||||
recommended action where applicable.
|
||||
"""
|
||||
52
src/mcp_cucm_axl/prompts/cucm_sql_help.py
Normal file
52
src/mcp_cucm_axl/prompts/cucm_sql_help.py
Normal file
@ -0,0 +1,52 @@
|
||||
"""Catch-all prompt for arbitrary CUCM SQL questions. Includes schema chunks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from ._common import render_schema_block
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..docs_loader import DocsIndex
|
||||
|
||||
|
||||
def render(docs: "DocsIndex | None", question: str) -> str:
|
||||
"""Generate a SQL-help prompt seeded with chunks relevant to the question."""
|
||||
keywords = [w for w in question.lower().split() if len(w) > 3][:8]
|
||||
if keywords:
|
||||
schema_block = render_schema_block(
|
||||
docs, keywords, max_chunks=5, max_chars_per_chunk=900
|
||||
)
|
||||
else:
|
||||
# Question has no substantive keywords — fall back to a generic
|
||||
# data-dictionary primer rather than trying to embed nothing.
|
||||
schema_block = render_schema_block(
|
||||
docs,
|
||||
["data dictionary", "informix", "schema"],
|
||||
max_chunks=3,
|
||||
max_chars_per_chunk=900,
|
||||
)
|
||||
|
||||
return f"""# CUCM SQL Question
|
||||
|
||||
The user asks: **{question}**
|
||||
|
||||
## How to approach this
|
||||
|
||||
1. If you don't recognize the relevant tables, call `axl_list_tables(pattern=...)`
|
||||
with a substring guess (e.g., "route%", "device%", "user%").
|
||||
2. Run `axl_describe_table(<table_name>)` for the candidate tables to see
|
||||
exact column names and types.
|
||||
3. If the schema chunks below already answer the question, draft the SQL
|
||||
directly. If not, also invoke the `cisco-docs` MCP server's `search_docs`
|
||||
tool with a relevant query (e.g., search_docs("data dictionary table for X")).
|
||||
4. Compose the SELECT, run it via `axl_sql(query=...)`.
|
||||
5. Summarize the result for the user — counts, anomalies, and what you'd
|
||||
recommend doing about them.
|
||||
|
||||
## Possibly relevant schema chunks
|
||||
|
||||
{schema_block}
|
||||
|
||||
Now answer the question.
|
||||
"""
|
||||
60
src/mcp_cucm_axl/prompts/investigate_pattern.py
Normal file
60
src/mcp_cucm_axl/prompts/investigate_pattern.py
Normal file
@ -0,0 +1,60 @@
|
||||
"""Deep-dive seed for one specific pattern. Schema chunks embedded."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from ._common import render_schema_block
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..docs_loader import DocsIndex
|
||||
|
||||
|
||||
_KEYWORDS = ["numplan", "transformation", "translation pattern", "route pattern"]
|
||||
|
||||
|
||||
def render(
|
||||
docs: "DocsIndex | None",
|
||||
pattern: str,
|
||||
partition: str | None = None,
|
||||
) -> str:
|
||||
"""Walk the user through a single pattern in detail."""
|
||||
schema_block = render_schema_block(
|
||||
docs, _KEYWORDS, max_chunks=4, max_chars_per_chunk=900
|
||||
)
|
||||
partition_clause = f" in partition `{partition}`" if partition else ""
|
||||
inspect_args = f"pattern={pattern!r}"
|
||||
if partition:
|
||||
inspect_args += f", partition={partition!r}"
|
||||
|
||||
return f"""# Investigate Pattern: `{pattern}`{partition_clause}
|
||||
|
||||
Walk the user through this pattern in detail.
|
||||
|
||||
## Suggested calls
|
||||
|
||||
1. `route_inspect_pattern({inspect_args})`
|
||||
— pattern detail, transformations, target route list/device, reverse CSS lookup
|
||||
2. `route_translation_chain(number=<sample number>)` — what other patterns
|
||||
would compete for matches if this pattern matched a real call
|
||||
3. If it's a route pattern with a route list target, follow with
|
||||
`route_lists_and_groups(name=<route list name>)`
|
||||
|
||||
## What to report
|
||||
|
||||
- **Type**: directory number / route / translation / hunt pilot / etc.
|
||||
- **Transformations applied**:
|
||||
- Called party transformation mask
|
||||
- Calling party transformation mask
|
||||
- Prefix digits
|
||||
- Digit discard instructions
|
||||
- **Routing target**: where does the call ultimately go?
|
||||
- **Who can reach it**: which CSSs include this pattern's partition? Which
|
||||
device-pool/phone classes use those CSSs?
|
||||
- **Anything anomalous**: missing description, undocumented transformations,
|
||||
patterns that shadow each other, etc.
|
||||
|
||||
## Reference: CUCM data dictionary
|
||||
|
||||
{schema_block}
|
||||
"""
|
||||
51
src/mcp_cucm_axl/prompts/route_plan_overview.py
Normal file
51
src/mcp_cucm_axl/prompts/route_plan_overview.py
Normal file
@ -0,0 +1,51 @@
|
||||
"""Snapshot of the cluster's routing setup, with schema reference embedded."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from ._common import ROUTE_KEYWORDS, render_schema_block
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..docs_loader import DocsIndex
|
||||
|
||||
|
||||
def render(docs: "DocsIndex | None") -> str:
|
||||
"""Use this when starting a fresh route-plan audit conversation."""
|
||||
schema_block = render_schema_block(docs, ROUTE_KEYWORDS, max_chunks=5)
|
||||
|
||||
return f"""# CUCM Route Plan Overview
|
||||
|
||||
You are auditing the routing configuration of a CUCM 15 cluster via the
|
||||
`mcp-cucm-axl` MCP server (read-only). Begin by gathering a high-level
|
||||
snapshot, then drill in where anything looks wrong or surprising.
|
||||
|
||||
## Suggested first calls (in order)
|
||||
|
||||
1. `axl_version()` — confirm cluster reachability + version
|
||||
2. `route_partitions()` — partition catalog with member counts
|
||||
3. `route_calling_search_spaces()` — CSS list with ordered partitions
|
||||
4. `route_patterns(kind="route")` — outbound route patterns
|
||||
5. `route_patterns(kind="translation")` — translation patterns
|
||||
6. `route_lists_and_groups()` — route list → route group → device chain
|
||||
7. `route_digit_discard_instructions()` — DDI catalog
|
||||
|
||||
## What to look for in your initial summary
|
||||
|
||||
- **Partition sprawl**: > ~30 partitions usually indicates accumulated
|
||||
legacy config. Note any with zero patterns or zero CSS membership.
|
||||
- **CSS-partition asymmetry**: partitions not reachable from any CSS are
|
||||
effectively dead.
|
||||
- **Pattern density**: which partitions hold the bulk of route/translation
|
||||
patterns? That's where the dial plan logic lives.
|
||||
- **Transformation patterns**: cgpn/cdpn masks and DDIs that look unusual
|
||||
or undocumented.
|
||||
- **Route list depth**: route lists with one route group are fine; with many,
|
||||
understand the failover order.
|
||||
|
||||
## Reference: CUCM data dictionary (route plan)
|
||||
|
||||
{schema_block}
|
||||
|
||||
Now run the calls above and produce a written audit summary.
|
||||
"""
|
||||
161
src/mcp_cucm_axl/prompts/sip_trunk_report.py
Normal file
161
src/mcp_cucm_axl/prompts/sip_trunk_report.py
Normal file
@ -0,0 +1,161 @@
|
||||
"""Comprehensive SIP trunk inventory: profiles, destinations, route-group
|
||||
membership, and findings template.
|
||||
|
||||
Implementation reference: `docs/query-patterns/sip-trunk-report.md`. The
|
||||
query patterns embedded below are the validated forms from that doc;
|
||||
update them in lockstep when the schema knowledge evolves.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from ._common import SIP_TRUNK_KEYWORDS, render_schema_block
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..docs_loader import DocsIndex
|
||||
|
||||
|
||||
def _esc(s: str) -> str:
|
||||
"""Inline string-literal escape — same convention as route_plan._esc.
|
||||
Single quotes get doubled for Informix; everything else passes through.
|
||||
"""
|
||||
return s.replace("'", "''")
|
||||
|
||||
|
||||
def render(docs: "DocsIndex | None", name_filter: str | None = None) -> str:
|
||||
"""Produce a SIP trunk inventory + findings prompt.
|
||||
|
||||
Args:
|
||||
name_filter: If given, narrow the inventory to trunks whose name
|
||||
matches the substring (case-sensitive LIKE). Otherwise
|
||||
includes all trunks.
|
||||
"""
|
||||
schema_block = render_schema_block(
|
||||
docs, SIP_TRUNK_KEYWORDS, max_chunks=4, max_chars_per_chunk=900
|
||||
)
|
||||
|
||||
# Build the LIKE clause for the name filter, if provided. The filter
|
||||
# value is escaped for SQL safety, then wrapped in `%`-bookends for
|
||||
# substring matching against device.name.
|
||||
if name_filter:
|
||||
safe = _esc(name_filter)
|
||||
name_clause = f"\n AND d.name LIKE '%{safe}%'"
|
||||
scope_note = f"narrowed to trunks matching `%{name_filter}%`"
|
||||
else:
|
||||
name_clause = ""
|
||||
scope_note = "all SIP trunks"
|
||||
|
||||
return f"""# CUCM SIP Trunk Report — {scope_note}
|
||||
|
||||
Produce a comprehensive SIP trunk inventory: profiles, destinations,
|
||||
downstream route-group membership, and a findings analysis. The queries
|
||||
below are the validated forms from `docs/query-patterns/sip-trunk-report.md`.
|
||||
|
||||
## Step 1 — Trunk inventory (one row per trunk)
|
||||
|
||||
```sql
|
||||
SELECT
|
||||
d.name AS trunk_name,
|
||||
d.description,
|
||||
sp.name AS sip_profile,
|
||||
css.name AS calling_search_space,
|
||||
dp.name AS device_pool,
|
||||
loc.name AS location,
|
||||
tsc.name AS preferred_codec,
|
||||
sd.requesturidomainname AS sip_domain,
|
||||
sd.isanonymous AS anon_caller_id,
|
||||
sd.preferrouteheaderdestination AS prefer_route_header,
|
||||
sd.acceptinboundrdnis AS accept_inbound_rdnis,
|
||||
sd.acceptoutboundrdnis AS accept_outbound_rdnis
|
||||
FROM device d
|
||||
JOIN typeclass tc ON d.tkclass = tc.enum
|
||||
JOIN sipdevice sd ON sd.fkdevice = d.pkid
|
||||
LEFT JOIN sipprofile sp ON d.fksipprofile = sp.pkid
|
||||
LEFT JOIN callingsearchspace css ON d.fkcallingsearchspace = css.pkid
|
||||
LEFT JOIN devicepool dp ON d.fkdevicepool = dp.pkid
|
||||
LEFT JOIN location loc ON d.fklocation = loc.pkid
|
||||
LEFT JOIN typesipcodec tsc ON sd.tksipcodec = tsc.enum
|
||||
WHERE tc.name = 'Trunk'{name_clause}
|
||||
ORDER BY d.name;
|
||||
```
|
||||
|
||||
Run via `axl_sql(query=<the SQL above>)`.
|
||||
|
||||
`anon_caller_id`, `prefer_route_header`, and the RDNIS flags return `'t'`
|
||||
or `'f'` (Informix bool encoding). Render with that in mind.
|
||||
|
||||
## Step 2 — Destinations (one row per IP/port; trunks can have multiple)
|
||||
|
||||
```sql
|
||||
SELECT
|
||||
d.name AS trunk_name,
|
||||
std.address,
|
||||
std.port,
|
||||
std.sortorder
|
||||
FROM siptrunkdestination std
|
||||
JOIN sipdevice sd ON std.fksipdevice = sd.pkid
|
||||
JOIN device d ON sd.fkdevice = d.pkid{name_clause.replace("d.name LIKE", "d.name LIKE")}
|
||||
ORDER BY d.name, std.sortorder;
|
||||
```
|
||||
|
||||
`address` may be an IP literal *or* a DNS name (Expressway-C trunks often
|
||||
use FQDNs). `port` defaults to 5060 (UDP/TCP) or 5061 (TLS).
|
||||
|
||||
## Step 3 — Route-group / route-list membership
|
||||
|
||||
Don't write raw SQL — use the existing tool:
|
||||
|
||||
```
|
||||
route_lists_and_groups()
|
||||
```
|
||||
|
||||
Filter the result for `route_groups[].devices[].class == "Trunk"` to find
|
||||
the (trunk → route group → route list) triples. Note: route lists with
|
||||
**route groups that have no static device members** resolve at call-time
|
||||
via the calling phone's device-pool `fkroutegroup_local` mapping (CUCM
|
||||
Standard Local Route Group feature). Trunks reachable only through Local
|
||||
Route Groups won't appear in the static result — call
|
||||
`route_device_pool_route_groups()` to enumerate those.
|
||||
|
||||
## Step 4 — Findings to call out
|
||||
|
||||
After the data is gathered, produce findings on each axis:
|
||||
|
||||
- **Single-point-of-failure trunks**: route groups with one trunk member
|
||||
where that group is the only path for a critical pattern (911,
|
||||
voicemail, fax). Cross-reference with `route_lists_and_groups()`.
|
||||
- **Profile sprawl vs. consolidation**: are N trunks using N different
|
||||
SIP profiles, or do most share a small number? Sprawl = harder to
|
||||
audit transport/timing settings consistently.
|
||||
- **CSS asymmetry**: PSTN-facing inbound trunks should typically have
|
||||
restrictive CSSs; internal-facing trunks (voicemail) should have
|
||||
permissive ones. Mismatches can cause one-way audio or routing failures.
|
||||
- **Codec heterogeneity**: most clusters standardize on G.711 µ-law.
|
||||
Trunks advertising G.722 or G.729 first warrant explanation.
|
||||
- **DNS-vs-IP destinations**: trunks using FQDNs depend on cluster DNS;
|
||||
flag if FQDN resolution adds an unsurfaced SPOF.
|
||||
- **Security posture**: trunks using `Non Secure SIP Trunk Profile` for
|
||||
carrier-facing connections are typical for premise SIP carriers but
|
||||
worth documenting as a deliberate choice.
|
||||
- **NULL primary CSS** on trunks is *not* automatically a finding —
|
||||
Expressway-C and InformaCast Fusion trunks legitimately have it
|
||||
(they don't originate generic outbound routing).
|
||||
|
||||
## Suggested follow-up calls
|
||||
|
||||
- `route_devices_using_css(css_name=<each unique trunk CSS>)` — what else
|
||||
uses the same CSS as a particular trunk? Identifies shared dependencies.
|
||||
- `route_inspect_pattern(pattern, partition)` — for each route pattern
|
||||
that targets a trunk-bearing route list, walk the call path.
|
||||
- `axl_sql("SELECT name, description FROM sipprofile")` — when multiple
|
||||
trunks share a SIP profile, look up the profile detail once.
|
||||
|
||||
## Reference: CUCM data dictionary (SIP trunk-related tables)
|
||||
|
||||
{schema_block}
|
||||
|
||||
Run the queries above, then produce a structured trunk-by-trunk report
|
||||
followed by the findings analysis. Use markdown tables for the inventory
|
||||
section; reserve prose for findings and recommendations.
|
||||
"""
|
||||
@ -331,128 +331,29 @@ def route_filters(name: str | None = None, include_members: bool = False) -> dic
|
||||
|
||||
# ====================================================================
|
||||
# Prompts — schema-grounded conversation seeds
|
||||
#
|
||||
# Bodies live in `mcp_cucm_axl.prompts.<name>`. The shims below are the
|
||||
# FastMCP registration surface; FastMCP introspects each shim's signature
|
||||
# to expose parameters to the LLM, so the parameter contract lives here.
|
||||
# Each shim is a thin pass-through to the corresponding render() function.
|
||||
# ====================================================================
|
||||
|
||||
_ROUTE_KEYWORDS = [
|
||||
"route plan", "route pattern", "translation pattern",
|
||||
"calling search space", "partition", "transformation",
|
||||
"digit discard", "numplan", "routepartition",
|
||||
]
|
||||
|
||||
_AUDIT_PROMPTS = {
|
||||
"route_plan_overview": [
|
||||
"route plan", "route pattern", "calling search space", "partition",
|
||||
],
|
||||
"translations": [
|
||||
"translation pattern", "called party transformation",
|
||||
"calling party transformation", "digit discard",
|
||||
],
|
||||
"css_partitions": [
|
||||
"calling search space", "partition", "css",
|
||||
],
|
||||
"transformations": [
|
||||
"called party transformation", "calling party transformation",
|
||||
"transformation mask", "prefix digits",
|
||||
],
|
||||
}
|
||||
from . import prompts as _prompts
|
||||
|
||||
|
||||
@mcp.prompt
|
||||
def route_plan_overview() -> str:
|
||||
"""Snapshot of the cluster's routing setup, with schema reference embedded.
|
||||
|
||||
Use this when you want to start a fresh route-plan audit conversation.
|
||||
"""
|
||||
chunks = []
|
||||
if _docs is not None:
|
||||
chunks = _docs.find(_ROUTE_KEYWORDS, max_chunks=5, max_chars_per_chunk=1000)
|
||||
schema_block = (
|
||||
_docs.format_chunks_for_prompt(chunks) if _docs else _docs_or_empty_msg()
|
||||
)
|
||||
|
||||
return f"""# CUCM Route Plan Overview
|
||||
|
||||
You are auditing the routing configuration of a CUCM 15 cluster via the
|
||||
`mcp-cucm-axl` MCP server (read-only). Begin by gathering a high-level
|
||||
snapshot, then drill in where anything looks wrong or surprising.
|
||||
|
||||
## Suggested first calls (in order)
|
||||
|
||||
1. `axl_version()` — confirm cluster reachability + version
|
||||
2. `route_partitions()` — partition catalog with member counts
|
||||
3. `route_calling_search_spaces()` — CSS list with ordered partitions
|
||||
4. `route_patterns(kind="route")` — outbound route patterns
|
||||
5. `route_patterns(kind="translation")` — translation patterns
|
||||
6. `route_lists_and_groups()` — route list → route group → device chain
|
||||
7. `route_digit_discard_instructions()` — DDI catalog
|
||||
|
||||
## What to look for in your initial summary
|
||||
|
||||
- **Partition sprawl**: > ~30 partitions usually indicates accumulated
|
||||
legacy config. Note any with zero patterns or zero CSS membership.
|
||||
- **CSS-partition asymmetry**: partitions not reachable from any CSS are
|
||||
effectively dead.
|
||||
- **Pattern density**: which partitions hold the bulk of route/translation
|
||||
patterns? That's where the dial plan logic lives.
|
||||
- **Transformation patterns**: cgpn/cdpn masks and DDIs that look unusual
|
||||
or undocumented.
|
||||
- **Route list depth**: route lists with one route group are fine; with many,
|
||||
understand the failover order.
|
||||
|
||||
## Reference: CUCM data dictionary (route plan)
|
||||
|
||||
{schema_block}
|
||||
|
||||
Now run the calls above and produce a written audit summary.
|
||||
Use this when starting a fresh route-plan audit conversation.
|
||||
"""
|
||||
return _prompts.route_plan_overview.render(_docs)
|
||||
|
||||
|
||||
@mcp.prompt
|
||||
def investigate_pattern(pattern: str, partition: str | None = None) -> str:
|
||||
"""Deep-dive seed for one specific pattern. Schema chunks embedded."""
|
||||
chunks = []
|
||||
if _docs is not None:
|
||||
chunks = _docs.find(
|
||||
["numplan", "transformation", "translation pattern", "route pattern"],
|
||||
max_chunks=4,
|
||||
max_chars_per_chunk=900,
|
||||
)
|
||||
schema_block = (
|
||||
_docs.format_chunks_for_prompt(chunks) if _docs else _docs_or_empty_msg()
|
||||
)
|
||||
partition_clause = f" in partition `{partition}`" if partition else ""
|
||||
|
||||
return f"""# Investigate Pattern: `{pattern}`{partition_clause}
|
||||
|
||||
Walk the user through this pattern in detail.
|
||||
|
||||
## Suggested calls
|
||||
|
||||
1. `route_inspect_pattern(pattern={pattern!r}{f', partition={partition!r}' if partition else ''})`
|
||||
— pattern detail, transformations, target route list/device, reverse CSS lookup
|
||||
2. `route_translation_chain(number=<sample number>)` — what other patterns
|
||||
would compete for matches if this pattern matched a real call
|
||||
3. If it's a route pattern with a route list target, follow with
|
||||
`route_lists_and_groups(name=<route list name>)`
|
||||
|
||||
## What to report
|
||||
|
||||
- **Type**: directory number / route / translation / hunt pilot / etc.
|
||||
- **Transformations applied**:
|
||||
- Called party transformation mask
|
||||
- Calling party transformation mask
|
||||
- Prefix digits
|
||||
- Digit discard instructions
|
||||
- **Routing target**: where does the call ultimately go?
|
||||
- **Who can reach it**: which CSSs include this pattern's partition? Which
|
||||
device-pool/phone classes use those CSSs?
|
||||
- **Anything anomalous**: missing description, undocumented transformations,
|
||||
patterns that shadow each other, etc.
|
||||
|
||||
## Reference: CUCM data dictionary
|
||||
|
||||
{schema_block}
|
||||
"""
|
||||
return _prompts.investigate_pattern.render(_docs, pattern, partition)
|
||||
|
||||
|
||||
@mcp.prompt
|
||||
@ -463,101 +364,25 @@ def audit_routing(focus: str = "full") -> str:
|
||||
focus: One of "full", "translations", "css_partitions", "transformations",
|
||||
"route_lists". Tunes which schema chunks get embedded.
|
||||
"""
|
||||
keyword_set = _AUDIT_PROMPTS.get(focus, _ROUTE_KEYWORDS)
|
||||
chunks = []
|
||||
if _docs is not None:
|
||||
chunks = _docs.find(keyword_set, max_chunks=6, max_chars_per_chunk=900)
|
||||
schema_block = (
|
||||
_docs.format_chunks_for_prompt(chunks) if _docs else _docs_or_empty_msg()
|
||||
)
|
||||
|
||||
return f"""# CUCM Routing Audit — focus: `{focus}`
|
||||
|
||||
Conduct a focused audit of the cluster's routing configuration. Goal: produce
|
||||
an actionable findings report — not just a description of the config.
|
||||
|
||||
## Audit checklist
|
||||
|
||||
### Partitions and access control
|
||||
- [ ] Are there partitions with zero patterns? (legacy/orphaned)
|
||||
- [ ] Are there partitions not referenced by any CSS? (unreachable)
|
||||
- [ ] Does the partition naming convention reflect actual scope?
|
||||
|
||||
### Calling Search Spaces
|
||||
- [ ] CSS-to-CSS comparison: any near-duplicates that differ only in order?
|
||||
- [ ] CSSs that include partitions in surprising orders (longest-match implications)
|
||||
- [ ] CSSs that are referenced by zero devices (use axl_sql against device table)
|
||||
|
||||
### Translation patterns
|
||||
- [ ] What does each translation pattern actually transform? Any with no
|
||||
transformation that exist purely for partition routing?
|
||||
- [ ] Calling-party transformations applied at translation: are they
|
||||
documented? Why is the calling number being rewritten?
|
||||
- [ ] Translation chains: do any translations route into partitions where
|
||||
another translation will match again? (chains can be intentional but
|
||||
obscure caller-ID and routing logic)
|
||||
|
||||
### Route patterns
|
||||
- [ ] Wildcard breadth: any `9.@` (NANPA at-sign) patterns? Are they intentional?
|
||||
- [ ] Block patterns: which patterns have `block_enabled` set? What are they
|
||||
blocking and why?
|
||||
- [ ] Patterns with no description — flag for documentation.
|
||||
|
||||
### Transformations (called party / calling party)
|
||||
- [ ] Digit discard instructions: which DDIs are referenced? Are any DDIs
|
||||
defined but never used?
|
||||
- [ ] Prefix-digits-out: any unusual prefixes (international, special service)?
|
||||
- [ ] Calling-party masks that hide internal extensions on outbound calls.
|
||||
|
||||
### Route lists and groups
|
||||
- [ ] Route lists with only one route group: simple, fine.
|
||||
- [ ] Route lists with many: walk the failover order, confirm it's intentional.
|
||||
- [ ] Route groups containing devices that are unregistered or disabled.
|
||||
|
||||
## Reference: CUCM data dictionary
|
||||
|
||||
{schema_block}
|
||||
|
||||
Run the relevant tool calls now and produce a structured findings report
|
||||
with category headers, observation, severity (info/warning/error), and
|
||||
recommended action where applicable.
|
||||
"""
|
||||
return _prompts.audit_routing.render(_docs, focus)
|
||||
|
||||
|
||||
@mcp.prompt
|
||||
def cucm_sql_help(question: str) -> str:
|
||||
"""Catch-all prompt for arbitrary CUCM SQL questions. Includes schema chunks."""
|
||||
keywords = [w for w in question.lower().split() if len(w) > 3][:8]
|
||||
chunks = []
|
||||
if _docs is not None and keywords:
|
||||
chunks = _docs.find(keywords, max_chunks=5, max_chars_per_chunk=900)
|
||||
schema_block = (
|
||||
_docs.format_chunks_for_prompt(chunks) if _docs else _docs_or_empty_msg()
|
||||
)
|
||||
return _prompts.cucm_sql_help.render(_docs, question)
|
||||
|
||||
return f"""# CUCM SQL Question
|
||||
|
||||
The user asks: **{question}**
|
||||
@mcp.prompt
|
||||
def sip_trunk_report(name_filter: str | None = None) -> str:
|
||||
"""Comprehensive SIP trunk inventory: profiles, destinations, route-group
|
||||
membership, and findings template.
|
||||
|
||||
## How to approach this
|
||||
|
||||
1. If you don't recognize the relevant tables, call `axl_list_tables(pattern=...)`
|
||||
with a substring guess (e.g., "route%", "device%", "user%").
|
||||
2. Run `axl_describe_table(<table_name>)` for the candidate tables to see
|
||||
exact column names and types.
|
||||
3. If the schema chunks below already answer the question, draft the SQL
|
||||
directly. If not, also invoke the `cisco-docs` MCP server's `search_docs`
|
||||
tool with a relevant query (e.g., search_docs("data dictionary table for X")).
|
||||
4. Compose the SELECT, run it via `axl_sql(query=...)`.
|
||||
5. Summarize the result for the user — counts, anomalies, and what you'd
|
||||
recommend doing about them.
|
||||
|
||||
## Possibly relevant schema chunks
|
||||
|
||||
{schema_block}
|
||||
|
||||
Now answer the question.
|
||||
Args:
|
||||
name_filter: Optional substring to narrow inventory to specific
|
||||
trunks (case-sensitive LIKE). Omit to include all SIP trunks.
|
||||
"""
|
||||
return _prompts.sip_trunk_report.render(_docs, name_filter)
|
||||
|
||||
|
||||
# ====================================================================
|
||||
|
||||
167
tests/test_prompts_package.py
Normal file
167
tests/test_prompts_package.py
Normal file
@ -0,0 +1,167 @@
|
||||
"""Tests for the extracted prompts package.
|
||||
|
||||
The render() functions are pure (they take docs as a parameter, no module
|
||||
globals), so they're trivially unit-testable. We verify each one renders
|
||||
without raising both with and without a docs index.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from mcp_cucm_axl import prompts
|
||||
from mcp_cucm_axl.docs_loader import DocsIndex
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_docs(tmp_path: Path) -> DocsIndex:
|
||||
chunks = [
|
||||
{
|
||||
"id": "cucm::v15::admin::Route-Plan::0",
|
||||
"text": "Route plan defines call routing.",
|
||||
"heading_path": ["Route Plan"],
|
||||
"source_path": str(tmp_path / "fake.md"),
|
||||
"product": "cucm",
|
||||
"version": "v15",
|
||||
"doc": "system-config-guide",
|
||||
},
|
||||
{
|
||||
"id": "cucm::v15::admin::SIP-Trunk::0",
|
||||
"text": "SIP trunks connect CUCM to other call control entities.",
|
||||
"heading_path": ["SIP Trunk Configuration"],
|
||||
"source_path": str(tmp_path / "fake.md"),
|
||||
"product": "cucm",
|
||||
"version": "v15",
|
||||
"doc": "system-config-guide",
|
||||
},
|
||||
]
|
||||
(tmp_path / "chunks.jsonl").write_text(
|
||||
"\n".join(json.dumps(c) for c in chunks)
|
||||
)
|
||||
(tmp_path / "index_meta.json").write_text(
|
||||
json.dumps({"model_name": "test", "embedding_dim": 384, "products": ["cucm"]})
|
||||
)
|
||||
idx = DocsIndex.load(tmp_path)
|
||||
assert idx is not None
|
||||
return idx
|
||||
|
||||
|
||||
# ---- route_plan_overview ----------------------------------------------------
|
||||
|
||||
def test_route_plan_overview_renders_with_docs(fake_docs):
|
||||
text = prompts.route_plan_overview.render(fake_docs)
|
||||
assert "Route Plan Overview" in text
|
||||
assert "axl_version" in text
|
||||
assert "route_partitions" in text
|
||||
|
||||
|
||||
def test_route_plan_overview_renders_without_docs():
|
||||
# Graceful degradation — no docs index should still produce a usable prompt
|
||||
text = prompts.route_plan_overview.render(None)
|
||||
assert "Route Plan Overview" in text
|
||||
assert "cisco-docs index is not loaded" in text
|
||||
|
||||
|
||||
# ---- investigate_pattern ---------------------------------------------------
|
||||
|
||||
def test_investigate_pattern_includes_pattern_in_output(fake_docs):
|
||||
text = prompts.investigate_pattern.render(fake_docs, "10.911", "CER911-PT")
|
||||
assert "10.911" in text
|
||||
assert "CER911-PT" in text
|
||||
|
||||
|
||||
def test_investigate_pattern_no_partition(fake_docs):
|
||||
text = prompts.investigate_pattern.render(fake_docs, "9.@")
|
||||
assert "9.@" in text
|
||||
# The partition clause should not appear when no partition is given
|
||||
assert "in partition" not in text
|
||||
|
||||
|
||||
# ---- audit_routing ----------------------------------------------------------
|
||||
|
||||
def test_audit_routing_default_focus(fake_docs):
|
||||
text = prompts.audit_routing.render(fake_docs)
|
||||
assert "focus: `full`" in text
|
||||
|
||||
|
||||
def test_audit_routing_custom_focus(fake_docs):
|
||||
text = prompts.audit_routing.render(fake_docs, "translations")
|
||||
assert "focus: `translations`" in text
|
||||
|
||||
|
||||
# ---- cucm_sql_help ----------------------------------------------------------
|
||||
|
||||
def test_cucm_sql_help_includes_question(fake_docs):
|
||||
q = "How do I find phones with no associated user?"
|
||||
text = prompts.cucm_sql_help.render(fake_docs, q)
|
||||
assert q in text
|
||||
assert "axl_describe_table" in text
|
||||
|
||||
|
||||
def test_cucm_sql_help_handles_empty_question(fake_docs):
|
||||
# No substantive keywords — must still render something useful
|
||||
text = prompts.cucm_sql_help.render(fake_docs, "x y z")
|
||||
assert "axl_list_tables" in text
|
||||
|
||||
|
||||
# ---- sip_trunk_report (NEW) ------------------------------------------------
|
||||
|
||||
def test_sip_trunk_report_includes_query_1(fake_docs):
|
||||
text = prompts.sip_trunk_report.render(fake_docs)
|
||||
# Query 1 essentials: device JOIN sipdevice, the trunk_name column,
|
||||
# and the WHERE filter on Trunk class
|
||||
assert "FROM device d" in text
|
||||
assert "JOIN sipdevice sd" in text
|
||||
assert "tc.name = 'Trunk'" in text
|
||||
|
||||
|
||||
def test_sip_trunk_report_includes_query_2_and_tools(fake_docs):
|
||||
text = prompts.sip_trunk_report.render(fake_docs)
|
||||
assert "siptrunkdestination" in text
|
||||
# And references the existing MCP tool for route group traversal
|
||||
assert "route_lists_and_groups()" in text
|
||||
|
||||
|
||||
def test_sip_trunk_report_with_name_filter_injects_safe_like(fake_docs):
|
||||
text = prompts.sip_trunk_report.render(fake_docs, "PSTN")
|
||||
# The filter should appear as a LIKE clause and the scope note should
|
||||
# reflect the narrowing
|
||||
assert "%PSTN%" in text
|
||||
assert "narrowed to trunks" in text
|
||||
|
||||
|
||||
def test_sip_trunk_report_name_filter_escaped(fake_docs):
|
||||
# SQL injection defense: a single-quote in the name_filter must be
|
||||
# doubled before going into the LIKE pattern
|
||||
text = prompts.sip_trunk_report.render(fake_docs, "O'Reilly")
|
||||
assert "O''Reilly" in text
|
||||
|
||||
|
||||
def test_sip_trunk_report_renders_without_docs():
|
||||
text = prompts.sip_trunk_report.render(None)
|
||||
assert "SIP Trunk Report" in text
|
||||
assert "cisco-docs index is not loaded" in text
|
||||
|
||||
|
||||
# ---- registration smoke test -----------------------------------------------
|
||||
|
||||
def test_all_prompts_registered_in_server():
|
||||
"""Confirm each prompt module's render() is wired through a FastMCP shim
|
||||
in server.py. Catches the case where a new module is added but the
|
||||
shim wasn't (the prompt would be invisible to the LLM)."""
|
||||
import asyncio
|
||||
from mcp_cucm_axl import server
|
||||
|
||||
async def _list():
|
||||
registered = await server.mcp.list_prompts()
|
||||
return {p.name for p in registered}
|
||||
|
||||
names = asyncio.run(_list())
|
||||
assert names == {
|
||||
"route_plan_overview",
|
||||
"investigate_pattern",
|
||||
"audit_routing",
|
||||
"cucm_sql_help",
|
||||
"sip_trunk_report",
|
||||
}, f"unexpected prompt set: {names}"
|
||||
Loading…
x
Reference in New Issue
Block a user