#!/usr/bin/env python3
"""MCP Server for JSON Skeleton Tool."""
import asyncio
import json
from typing import Dict, Any
from pathlib import Path
from mcp.server import Server
from mcp.types import (
Tool,
TextContent
)
from .skeleton_generator import SkeletonGenerator
# Initialize server
server = Server("json-skeleton-mcp")
skeleton_generator = SkeletonGenerator()
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="json_skeleton",
description="Create a lightweight JSON skeleton that preserves structure with truncated values and deduplicated arrays. Useful when encountering 'File content exceeds maximum allowed size' errors with large JSON files.",
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Path to the JSON file to process"
},
"max_length": {
"type": "integer",
"description": "Maximum length for string values (default: 200)",
"default": 200
},
"type_only": {
"type": "boolean",
"description": "Return only value types instead of values. Most compact output. (default: false)",
"default": False
}
},
"required": ["file_path"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> list[TextContent]:
"""Handle tool calls."""
try:
if name == "json_skeleton":
file_path = arguments.get("file_path")
if not file_path:
return [TextContent(type="text", text="Error: file_path is required")]
max_length = arguments.get("max_length", 200)
type_only = arguments.get("type_only", False)
result = skeleton_generator.process_file(file_path, max_length=max_length, type_only=type_only)
# Format the output - just the skeleton
output = json.dumps(result['skeleton'], indent=2, ensure_ascii=False)
return [TextContent(type="text", text=output)]
else:
return [TextContent(type="text", text=f"Error: Unknown tool '{name}'")]
except FileNotFoundError as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
except ValueError as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
except Exception as e:
# Log error to stderr for debugging
import sys
print(f"Error processing file: {str(e)}", file=sys.stderr)
return [TextContent(type="text", text=f"Error: Failed to process file - {str(e)}")]
async def run():
"""Run the MCP server."""
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
def main():
"""Main entry point."""
try:
asyncio.run(run())
except KeyboardInterrupt:
pass
except Exception as e:
print(f"Server error: {e}")
raise
if __name__ == "__main__":
main()
