mcp_server.py•9.93 kB
#!/usr/bin/env python3
"""
JSON Schema Validator MCP Server
This MCP server provides tools for validating JSON against JSON Schema Draft 2020-12
with external reference support.
"""
import asyncio
import logging
from typing import Any, Dict, List, Optional
from mcp.server import Server
from mcp.server.lowlevel import NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
CallToolResult,
TextContent,
ContentBlock,
EmptyResult,
Tool,
)
from tools.JSONSchemaValidator import JSONSchemaValidator
from utils.DataManager import DataManager
from tools.SchemaValidator import SchemaValidator
from handlers import (
handle_validate_json_schema,
handle_validate_json_from_collections,
handle_get_validation_info,
handle_add_update_schema,
handle_delete_schema,
handle_get_schema,
handle_list_schemas,
handle_generate_schema
)
# Configure logging
logging.basicConfig(level=logging.INFO,handlers=[logging.FileHandler("logs/mcp_server.log", encoding='utf-8')])
logger = logging.getLogger(__name__)
# Create server instance
server = Server("json-schema-validator")
# Global validator and data manager instances
validator = JSONSchemaValidator()
data_manager = DataManager()
schema_validator = SchemaValidator()
@server.list_tools()
async def handle_list_tools() -> List[Tool]:
"""
Lists the available tools in the MCP server.
"""
return [
Tool(
name="validate_json_schema",
description="Validates JSON against a JSON Schema Draft 2020-12 with external reference support",
inputSchema={
"type": "object",
"properties": {
"json_data": {
"type": ["object", "string"],
"description": "JSON data to validate (object or JSON string)"
},
"schema": {
"type": ["object", "string"],
"description": "JSON Schema to use for validation (object or JSON string)"
},
"strict": {
"type": "boolean",
"description": "If True, uses strict validation (optional, default: True)",
"default": True
}
},
"required": ["json_data", "schema"]
}
),
Tool(
name="validate_json_from_collections",
description="Validates JSON using schemas from data collections (.schemas)",
inputSchema={
"type": "object",
"properties": {
"json_data": {
"type": ["object", "string"],
"description": "JSON data to validate (object or JSON string)"
},
"schema_id": {
"type": "string",
"description": "Schema ID in the .schemas collection"
},
"strict": {
"type": "boolean",
"description": "If True, uses strict validation (optional, default: True)",
"default": True
}
},
"required": ["json_data", "schema_id"]
}
),
Tool(
name="get_validation_info",
description="Gets information about the validator and supported capabilities",
inputSchema={
"type": "object",
"properties": {},
"additionalProperties": False
}
),
Tool(
name="add_update_schema",
description="Adds or updates a JSON Schema in the collections",
inputSchema={
"type": "object",
"properties": {
"schema_id": {
"type": "string",
"description": "Schema ID (Linux path format, must end in .json)"
},
"schema_content": {
"type": "string",
"description": "JSON Schema content as string"
},
"update_if_exists": {
"type": "boolean",
"description": "If True, updates the schema if it already exists (default: False)",
"default": False
}
},
"required": ["schema_id", "schema_content"]
}
),
Tool(
name="delete_schema",
description="Deletes a JSON Schema from the collections",
inputSchema={
"type": "object",
"properties": {
"schema_id": {
"type": "string",
"description": "Schema ID to delete (Linux path format, must end in .json)"
},
"confirm_deletion": {
"type": "boolean",
"description": "Explicit confirmation to delete the schema (default: False)",
"default": False
}
},
"required": ["schema_id", "confirm_deletion"]
}
),
Tool(
name="get_schema",
description="Gets the content of a JSON Schema from the collections",
inputSchema={
"type": "object",
"properties": {
"schema_id": {
"type": "string",
"description": "Schema ID to retrieve (Linux path format, must end in .json)"
}
},
"required": ["schema_id"]
}
),
Tool(
name="list_schemas",
description="Lists all available schema IDs in the .schemas collection",
inputSchema={
"type": "object",
"properties": {},
"additionalProperties": False
}
),
Tool(
name="generate_schema",
description="Generates a JSON Schema from JSON data and saves it to the collections",
inputSchema={
"type": "object",
"properties": {
"schema_id": {
"type": "string",
"description": "Schema ID to generate (Linux path format, must end in .json)"
},
"json_data": {
"type": ["object", "string"],
"description": "JSON data from which to generate the schema (object or JSON string)"
},
"null_handling": {
"type": "string",
"enum": ["allow", "ignore", "strict"],
"description": "Strategy for handling null values: 'allow' (allows nulls), 'ignore' (ignores null fields), 'strict' (treats nulls as strings)",
"default": "allow"
}
},
"required": ["schema_id", "json_data"]
}
)
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any] | None = None) -> List[ContentBlock]:
"""
Handles calls to MCP server tools.
"""
try:
arguments = arguments or {}
logger.info(f"Starting call handler...{name}")
if name == "validate_json_schema":
result = await handle_validate_json_schema(arguments, validator, data_manager)
elif name == "validate_json_from_collections":
result = await handle_validate_json_from_collections(arguments, validator, data_manager)
elif name == "get_validation_info":
result = await handle_get_validation_info(arguments, validator, data_manager)
elif name == "add_update_schema":
result = await handle_add_update_schema(arguments, validator, data_manager, schema_validator)
elif name == "delete_schema":
result = await handle_delete_schema(arguments, validator, data_manager, schema_validator)
elif name == "get_schema":
result = await handle_get_schema(arguments, validator, data_manager, schema_validator)
elif name == "list_schemas":
result = await handle_list_schemas(arguments, validator, data_manager, schema_validator)
elif name == "generate_schema":
result = await handle_generate_schema(arguments, validator, data_manager, schema_validator)
else:
raise ValueError(f"Unknown tool: {name}")
logger.info(f"Call handler completed: {result.content[0]}")
return [result.content[0]]
except Exception as e:
logger.error(f"Error in tool {name}: {e}")
return [TextContent(
type="text",
text=f"Error executing {name}: {str(e)}"
)]
async def main():
"""
Main function to run the MCP server.
"""
logger.info("Starting JSON Schema Validator MCP Server...")
# Configure initialization options
init_options = InitializationOptions(
server_name="json-schema-validator",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
)
# Run server using stdio
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
init_options
)
if __name__ == "__main__":
asyncio.run(main())