Skip to content
231 changes: 85 additions & 146 deletions scripts/mcp-list.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
#!/usr/bin/env python3
"""Generate canonical tool listings from MCP server code.
"""Generate canonical tool and resource listings from MCP server code.

This script inspects the actual MCP server implementation to generate
authoritative tool listings, eliminating manual maintenance and drift.
authoritative tool and resource listings, eliminating manual maintenance and drift.
Resources are distinguished from tools with a 'type' column in the CSV output.
"""

import csv
Expand All @@ -22,6 +23,7 @@
sys.path.insert(0, str(src_dir))

from quilt_mcp.utils import create_configured_server
from quilt_mcp.resources import create_default_registry

async def extract_tool_metadata(server) -> List[Dict[str, Any]]:
"""Extract comprehensive metadata from all registered tools."""
Expand Down Expand Up @@ -50,6 +52,7 @@ async def extract_tool_metadata(server) -> List[Dict[str, Any]]:
signature_str = f"{tool_name}{sig}"

tools.append({
"type": "tool",
"name": tool_name,
"module": short_module,
"signature": signature_str,
Expand All @@ -63,155 +66,92 @@ async def extract_tool_metadata(server) -> List[Dict[str, Any]]:
tools.sort(key=lambda x: (x["module"], x["name"]))
return tools

def generate_csv_output(tools: List[Dict[str, Any]], output_file: str):
"""Generate CSV output matching current format."""
async def extract_resource_metadata() -> List[Dict[str, Any]]:
"""Extract comprehensive metadata from all registered resources."""
resources = []

registry = create_default_registry()
for uri_pattern in registry.list_resources():
resource = registry.get_resource(uri_pattern)

# Get resource class information
resource_class = resource.__class__
module = inspect.getmodule(resource_class)
module_name = module.__name__ if module else "unknown"

# Extract module short name (last component)
if module_name.startswith("quilt_mcp.resources."):
short_module = module_name.replace("quilt_mcp.resources.", "")
else:
short_module = module_name

# Get class docstring
doc = inspect.getdoc(resource_class) or "No description available"

# Build signature string showing the URI pattern
signature_str = f"{resource_class.__name__}(uri='{uri_pattern}')"

# Check if list_items method is async
is_async = inspect.iscoroutinefunction(resource.list_items)

resources.append({
"type": "resource",
"name": uri_pattern,
"module": short_module,
"signature": signature_str,
"description": doc.split('\n')[0], # First line only
"is_async": is_async,
"full_module_path": module_name,
"handler_class": resource_class.__name__
})

# Sort by URI pattern for consistent ordering
resources.sort(key=lambda x: x["name"])
return resources

def generate_csv_output(items: List[Dict[str, Any]], output_file: str):
"""Generate CSV output for tools and resources with type column."""
with open(output_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow([
"module", "function_name", "signature", "description",
"type", "module", "function_name", "signature", "description",
"is_async", "full_module_path"
])

for tool in tools:
for item in items:
writer.writerow([
tool["module"],
tool["name"],
tool["signature"],
tool["description"],
str(tool["is_async"]),
tool["full_module_path"]
item["type"],
item["module"],
item["name"],
item["signature"],
item["description"],
str(item["is_async"]),
item["full_module_path"]
])

def generate_json_output(tools: List[Dict[str, Any]], output_file: str):
def generate_json_output(items: List[Dict[str, Any]], output_file: str):
"""Generate structured JSON output for tooling."""
tools = [item for item in items if item["type"] == "tool"]
resources = [item for item in items if item["type"] == "resource"]

output = {
"metadata": {
"generated_by": "scripts/generate_canonical_tools.py",
"generated_by": "scripts/mcp-list.py",
"tool_count": len(tools),
"modules": list(set(tool["module"] for tool in tools))
"resource_count": len(resources),
"total_count": len(items),
"modules": list(set(item["module"] for item in items))
},
"tools": tools
"tools": tools,
"resources": resources
}

with open(output_file, 'w', encoding='utf-8') as f:
json.dump(output, f, indent=2, ensure_ascii=False)

def identify_overlapping_tools(_tools: List[Dict[str, Any]]) -> Dict[str, List[str]]:
"""Identify tools with overlapping functionality that should be consolidated."""
overlaps = {}

# Package creation tools - NOW CONSOLIDATED
package_creation = [
"create_package", # unified_package module - primary interface
"package_create_from_s3" # s3_package module - specialized S3 bulk processing
]
overlaps["Package Creation"] = package_creation

# Catalog/URL generation tools - REDUNDANT
catalog_tools = [
"catalog_url", # auth module
"catalog_uri" # auth module
]
overlaps["Catalog URLs"] = catalog_tools

# Metadata template tools - PARTIAL OVERLAP
metadata_tools = [
"get_metadata_template", # metadata_templates module
"create_metadata_from_template" # metadata_examples module
]
overlaps["Metadata Templates"] = metadata_tools

# Search tools - CONSOLIDATION NEEDED
search_tools = [
"packages_search", # packages module - package-specific
"bucket_objects_search", # buckets module - S3-specific
"unified_search" # search module - unified interface
]
overlaps["Search Functions"] = search_tools

# Tabulator admin overlap - DUPLICATE FUNCTIONALITY
tabulator_admin = [
"tabulator_open_query_status", # tabulator module
"tabulator_open_query_toggle", # tabulator module
"admin_tabulator_open_query_get", # governance module
"admin_tabulator_open_query_set" # governance module
]
overlaps["Tabulator Admin"] = tabulator_admin

return overlaps

def generate_consolidation_report(_tools: List[Dict[str, Any]], output_file: str):
"""Generate detailed consolidation recommendations."""

report = {
"breaking_changes_required": True,
"backward_compatibility": "DEPRECATED - Will break existing clients",
"consolidation_plan": {}
}

# Package Creation Consolidation - COMPLETED
report["consolidation_plan"]["package_creation"] = {
"action": "COMPLETED",
"keep": ["create_package", "package_create_from_s3"],
"removed": ["package_create", "package_update", "package_update_metadata"],
"rationale": "create_package is now the unified primary interface with all functionality",
"current_api": {
"create_package": "Primary interface - handles all package creation scenarios",
"package_create_from_s3": "Specialized tool for S3 bulk processing with organization"
}
}

# Search Consolidation
report["consolidation_plan"]["search"] = {
"action": "BREAK_COMPATIBILITY",
"keep": "unified_search",
"deprecate": ["packages_search", "bucket_objects_search"],
"rationale": "unified_search handles all search scenarios with backend selection",
"migration": {
"packages_search": "Replace with unified_search(scope='catalog')",
"bucket_objects_search": "Replace with unified_search(scope='bucket', target=bucket)"
}
}

# URL Generation Consolidation
report["consolidation_plan"]["url_generation"] = {
"action": "BREAK_COMPATIBILITY",
"keep": "catalog_url",
"deprecate": ["catalog_uri"],
"rationale": "catalog_url covers all URL generation needs",
"migration": {
"catalog_uri": "Replace with catalog_url() - URIs are legacy"
}
}

# Tabulator Admin Consolidation
report["consolidation_plan"]["tabulator_admin"] = {
"action": "BREAK_COMPATIBILITY",
"keep": ["admin_tabulator_open_query_get", "admin_tabulator_open_query_set"],
"deprecate": ["tabulator_open_query_status", "tabulator_open_query_toggle"],
"rationale": "Admin tools provide proper permissions model",
"migration": {
"tabulator_open_query_status": "Replace with admin_tabulator_open_query_get",
"tabulator_open_query_toggle": "Replace with admin_tabulator_open_query_set"
}
}

# Documentation Cleanup
report["documentation_cleanup"] = {
"action": "REGENERATE_FROM_CODE",
"current_issues": [
"docs/api/TOOLS.md manually maintained - causes drift",
"CSV file manually updated - inconsistent with code",
"Tool descriptions in docs don't match actual docstrings"
],
"solution": "Auto-generate all documentation from server introspection"
}

with open(output_file, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, ensure_ascii=False)

async def main():
"""Generate all canonical tool listings."""
"""Generate all canonical tool and resource listings."""
print("🔍 Extracting tools from MCP server...")

# Create server instance to introspect tools
Expand All @@ -220,33 +160,32 @@ async def main():

print(f"📊 Found {len(tools)} tools across {len(set(tool['module'] for tool in tools))} modules")

print("🔍 Extracting resources from MCP registry...")
resources = await extract_resource_metadata()

print(f"📊 Found {len(resources)} resources across {len(set(resource['module'] for resource in resources))} modules")

# Combine tools and resources
all_items = tools + resources

# Generate outputs
output_dir = Path(__file__).parent.parent
tests_fixtures_dir = output_dir / "tests" / "fixtures"

print("📝 Generating CSV output...")
generate_csv_output(tools, str(tests_fixtures_dir / "mcp-list.csv"))
generate_csv_output(all_items, str(tests_fixtures_dir / "mcp-list.csv"))

print("📋 Generating JSON metadata...")
generate_json_output(tools, str(output_dir / "build" / "tools_metadata.json"))

print("⚠️ Generating consolidation report...")
generate_consolidation_report(tools, str(output_dir / "build" / "consolidation_report.json"))

# Print summary
overlaps = identify_overlapping_tools(tools)
print("\n🚨 OVERLAPPING TOOLS IDENTIFIED:")
for category, tool_list in overlaps.items():
print(f" {category}: {len(tool_list)} tools")
for tool in tool_list:
print(f" - {tool}")
print()
generate_json_output(all_items, str(output_dir / "build" / "tools_metadata.json"))

print("✅ Canonical tool listings generated!")
print("✅ Canonical tool and resource listings generated!")
print("📂 Files created:")
print(" - tests/fixtures/mcp-list.csv")
print(" - build/tools_metadata.json")
print(" - build/consolidation_report.json")
print(f"📈 Summary:")
print(f" - {len(tools)} tools")
print(f" - {len(resources)} resources")
print(f" - {len(all_items)} total items")

if __name__ == "__main__":
import asyncio
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,10 @@
import pytest
from quilt_mcp.tools.catalog import catalog_status
from quilt_mcp.tools.packages import (
packages_list,
package_browse,
package_contents_search,
)
from quilt_mcp.tools.search import catalog_search


@pytest.mark.aws
Expand All @@ -18,14 +18,14 @@ def test_quilt_tools():
result = catalog_status()
assert isinstance(result, dict)

# Basic listing call should return dict (mocked in unit runs)
# Basic catalog search should return dict (replaces packages_list)
try:
pkgs = packages_list()
pkgs = catalog_search(query="*", scope="catalog", limit=10)
assert isinstance(pkgs, dict)
except Exception as e:
if "AccessDenied" in str(e) or "S3NoValidClientError" in str(e):
# Expected in environments without proper AWS permissions
pkgs = {"packages": [], "error": "Access denied"}
pkgs = {"results": [], "error": "Access denied"}
else:
raise

Expand Down
23 changes: 12 additions & 11 deletions tests/integration/test_local.py → scripts/tests/test_tools.py
100644 → 100755
Original file line number Diff line number Diff line change
@@ -1,18 +1,19 @@
#!/usr/bin/env python3
"""
Test script for the Quilt MCP server
Syntactic smoke test for Quilt MCP server tools.

This is a purely syntactic test that verifies function signatures and basic
return types without requiring AWS credentials or network access.
"""

import asyncio
import pytest
from quilt_mcp.tools import catalog as auth_tools
from quilt_mcp.tools.packages import packages_list, package_browse
from quilt_mcp.tools.packages import package_browse
from quilt_mcp.tools.search import catalog_search


@pytest.mark.asyncio
@pytest.mark.aws
async def test_quilt_tools():
"""Basic smoke test against local tool functions (no network)."""
"""Basic smoke test against tool function signatures (no network/AWS required)."""
print("Testing Quilt MCP Server Tools...")
print("=" * 50)

Expand All @@ -21,15 +22,15 @@ async def test_quilt_tools():
result = auth_tools.catalog_status()
assert isinstance(result, dict)

# Test packages_list signature (dry call with default registry)
print("\n2. Testing packages_list (dry):")
# Test catalog_search (replaces packages_list)
print("\n2. Testing catalog_search (dry):")
try:
response = packages_list()
response = catalog_search(query="*", scope="catalog", limit=10)
assert isinstance(response, dict)
except Exception as e:
if "AccessDenied" in str(e) or "S3NoValidClientError" in str(e):
print(f"Skipped packages_list due to access denied: {e}")
response = {"packages": [], "error": "Access denied"}
print(f"Skipped catalog_search due to access denied: {e}")
response = {"results": [], "error": "Access denied"}
else:
raise

Expand Down
2 changes: 0 additions & 2 deletions src/quilt_mcp/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,6 @@
package_browse,
package_contents_search,
package_diff,
packages_list,
)
from .tools.tabulator import (
tabulator_table_create,
Expand Down Expand Up @@ -124,7 +123,6 @@
"package_delete",
"package_diff",
"package_validate",
"packages_list",
"quick_start",
"tabulator_table_create",
"tabulator_table_delete",
Expand Down
Loading