58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
# web/mcp_introspection.py
|
|
"""Introspect MCP tools from the FastMCP instance."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def list_mcp_tools() -> list[dict[str, Any]]:
|
|
"""List all registered MCP tools with metadata."""
|
|
from legal_mcp.server import mcp
|
|
|
|
tools = await mcp.list_tools()
|
|
out: list[dict[str, Any]] = []
|
|
for t in tools:
|
|
# Resolve underlying callable for source location
|
|
fn = _resolve_callable(t.name)
|
|
source_location = ""
|
|
module = ""
|
|
if fn is not None:
|
|
try:
|
|
file = inspect.getfile(fn)
|
|
_, line = inspect.getsourcelines(fn)
|
|
source_location = f"{file}:{line}"
|
|
module = fn.__module__
|
|
except Exception as e:
|
|
logger.debug(
|
|
"tool_source_resolution_failed name=%s err=%s", t.name, e,
|
|
)
|
|
out.append({
|
|
"name": t.name,
|
|
"description": t.description or "",
|
|
"params_schema": getattr(t, "inputSchema", None),
|
|
"module": module,
|
|
"source_location": source_location,
|
|
})
|
|
return sorted(out, key=lambda r: (r["module"], r["name"]))
|
|
|
|
|
|
def _resolve_callable(tool_name: str):
|
|
"""Find the python function backing a registered tool name."""
|
|
from legal_mcp.tools import (
|
|
cases, documents, drafting, precedent_library,
|
|
precedents, search, workflow,
|
|
)
|
|
for mod in (
|
|
cases, documents, drafting, precedent_library,
|
|
precedents, search, workflow,
|
|
):
|
|
fn = getattr(mod, tool_name, None)
|
|
if callable(fn):
|
|
return fn
|
|
return None
|