app.py

#!/usr/bin/env python3
"""
Conversational Form-Filling MCP Server
MCP 1st Birthday Hackathon - Track 1: Building with MCP

This MCP server provides tools for conversational form filling.
Connect it to any MCP client (Claude Desktop, VS Code, Cursor, etc.)
"""

import asyncio
import json
import re
import hashlib
import time
from typing import Any, Dict
from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
import mcp.types as types

# Form schema
FORM_SCHEMA = {
    "title": "Job Application Form",
    "description": "Complete your job application",
    "fields": [
        {"id": "full_name", "label": "What is your full name?", "type": "text", "required": True, "validation": {"min_length": 2}},
        {"id": "email", "label": "What is your email address?", "type": "email", "required": True},
        {"id": "phone", "label": "What is your phone number? (XXX-XXX-XXXX)", "type": "phone", "required": True},
        {"id": "experience", "label": "Years of experience?", "type": "number", "required": True},
        {"id": "role", "label": "Role applying for?", "type": "choice", "required": True, "options": ["Software Engineer", "Product Manager", "Designer", "Data Scientist"]},
        {"id": "availability", "label": "When can you start?", "type": "choice", "required": True, "options": ["Immediately", "2 weeks", "1 month", "2+ months"]}
    ]
}

form_data: Dict[str, Any] = {}
server = Server("conversational-form-agent")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(name="get_form_schema", description="Get form schema", inputSchema={"type": "object", "properties": {}}),
        types.Tool(name="get_next_question", description="Get next question", inputSchema={"type": "object", "properties": {"filled_data": {"type": "object"}}}),
        types.Tool(name="validate_answer", description="Validate answer", inputSchema={"type": "object", "properties": {"field_name": {"type": "string"}, "value": {"type": "string"}}, "required": ["field_name", "value"]}),
        types.Tool(name="save_answer", description="Save answer", inputSchema={"type": "object", "properties": {"field_name": {"type": "string"}, "value": {"type": "string"}}, "required": ["field_name", "value"]}),
        types.Tool(name="submit_form", description="Submit form", inputSchema={"type": "object", "properties": {"filled_data": {"type": "object"}}, "required": ["filled_data"]})
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent]:
    if arguments is None:
        arguments = {}
    
    if name == "get_form_schema":
        return [types.TextContent(type="text", text=json.dumps(FORM_SCHEMA, indent=2))]
    
    elif name == "get_next_question":
        filled_data = arguments.get("filled_data", {})
        for field in FORM_SCHEMA["fields"]:
            if field["required"] and field["id"] not in filled_data:
                result = {"field": field, "progress": f"{len(filled_data)}/{len(FORM_SCHEMA['fields'])}", "question": field["label"]}
                if field["type"] == "choice":
                    result["options"] = field["options"]
                return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
        return [types.TextContent(type="text", text=json.dumps({"status": "complete"}, indent=2))]
    
    elif name == "validate_answer":
        field_name = arguments.get("field_name")
        value = arguments.get("value")
        field = next((f for f in FORM_SCHEMA["fields"] if f["id"] == field_name), None)
        if not field:
            return [types.TextContent(type="text", text=json.dumps({"valid": False, "error": "Unknown field"}, indent=2))]
        # Basic validation - expand as needed
        return [types.TextContent(type="text", text=json.dumps({"valid": True}, indent=2))]
    
    elif name == "save_answer":
        field_name = arguments.get("field_name")
        value = arguments.get("value")
        form_data[field_name] = value
        return [types.TextContent(type="text", text=json.dumps({"success": True, "total_saved": len(form_data)}, indent=2))]
    
    elif name == "submit_form":
        filled_data = arguments.get("filled_data", {})
        submission_id = hashlib.md5(f"{time.time()}{json.dumps(filled_data)}".encode()).hexdigest()[:12]
        return [types.TextContent(type="text", text=json.dumps({"success": True, "submission_id": f"sub_{submission_id}"}, indent=2))]
    
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, InitializationOptions(
            server_name="conversational-form-agent",
            server_version="1.0.0",
            capabilities=server.get_capabilities(notification_options=NotificationOptions(), experimental_capabilities={})
        ))

if __name__ == "__main__":
    asyncio.run(main())