"""
Validation Handlers for JSON Schema MCP Server
This module contains handlers for JSON validation operations.
"""
import logging
from typing import Any, Dict
from mcp.types import CallToolResult, TextContent
logger = logging.getLogger(__name__)
async def handle_validate_json_schema(arguments: Dict[str, Any], validator, data_manager) -> CallToolResult:
"""
Handles direct JSON validation against a schema.
"""
json_data = arguments.get("json_data")
schema = arguments.get("schema")
strict = arguments.get("strict", True)
if not json_data or not schema:
return CallToolResult(
content=[
TextContent(
type="text",
text="Error: 'json_data' and 'schema' parameters are required"
)
]
)
# Validate using JSONSchemaValidator
result = validator.validate_json(json_data, schema, strict=strict)
# Format result for MCP
if result["valid"]:
status = "VALID"
details = "JSON data is valid according to the provided schema."
else:
status = "INVALID"
details = f"Found {result['error_count']} errors:\n"
for i, error in enumerate(result["errors"], 1):
details += f" {i}. {error}\n"
response_text = f"""JSON Schema Validation Result
{status}
{details}
Summary:
- Errors found: {result['error_count']}
- Strict validation: {strict}
"""
return CallToolResult(
content=[
TextContent(
type="text",
text=response_text
)
]
)
async def handle_validate_json_from_collections(arguments: Dict[str, Any], validator, data_manager) -> CallToolResult:
"""
Handles validation using schemas from collections.
"""
json_data = arguments.get("json_data")
schema_id = arguments.get("schema_id")
strict = arguments.get("strict", True)
if not json_data or not schema_id:
return CallToolResult(
content=[
TextContent(
type="text",
text="Error: 'json_data' and 'schema_id' parameters are required"
)
]
)
try:
# Load schema from collection using DataManager
schema = data_manager.load_data(".schemas", schema_id)
# Validate using JSONSchemaValidator
result = validator.validate_json(json_data, schema, strict=strict)
# Format result for MCP
if result["valid"]:
status = "VALID"
details = f"JSON data is valid according to schema '{schema_id}'."
else:
status = "INVALID"
details = f"Found {result['error_count']} errors:\n"
for i, error in enumerate(result["errors"], 1):
details += f" {i}. {error}\n"
response_text = f"""JSON Schema Validation Result
{status}
Schema used: {schema_id}
{details}
Summary:
- Errors found: {result['error_count']}
- Strict validation: {strict}
"""
return CallToolResult(
content=[
TextContent(
type="text",
text=response_text
)
]
)
except Exception as e:
return CallToolResult(
content=[
TextContent(
type="text",
text=f"Error loading schema '{schema_id}': {str(e)}"
)
]
)
async def handle_get_validation_info(arguments: Dict[str, Any], validator, data_manager) -> CallToolResult:
"""
Provides information about the validator and capabilities.
"""
info_text = """JSON Schema MCP Validator Information
Features:
- JSON Schema Draft 2020-12 compatible
- External reference support ($ref)
- Automatic internal reference resolution (#/$defs/name)
- Schema caching for better performance
- Detailed error messages with specific paths
Available tools:
1. validate_json_schema - Direct validation with provided schema
2. validate_json_from_collections - Validation using schemas from collections
3. get_validation_info - This information
4. add_update_schema - Add or update schemas in collections
5. delete_schema - Delete schemas from collections
6. get_schema - Get schema content from collections
7. list_schemas - List all available schema IDs
"""
return CallToolResult(
content=[
TextContent(
type="text",
text=info_text,
)
]
)
