app.py

import gradio as gr
import httpx
import json
import re

BASE_URL = "https://nelsonandre.outsystemscloud.com/MCPServer/rest/MCP"
MAX_STRING_LENGTH = 500
DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")


def validate_employee_id(employee_id) -> int:
    """Validate and convert employee ID to a positive integer."""
    try:
        eid = int(employee_id)
        if eid <= 0:
            raise ValueError("Employee ID must be a positive integer")
        return eid
    except (TypeError, ValueError) as e:
        raise ValueError(f"Invalid employee ID: {e}")


def validate_string(value: str, field_name: str, required: bool = False, max_length: int = MAX_STRING_LENGTH) -> str:
    """Validate and sanitize string input."""
    if value is None:
        value = ""
    value = str(value).strip()
    if required and not value:
        raise ValueError(f"{field_name} is required")
    if len(value) > max_length:
        raise ValueError(f"{field_name} exceeds maximum length of {max_length}")
    return value


def validate_date(value: str, field_name: str) -> str:
    """Validate date format (YYYY-MM-DD)."""
    if not value:
        return ""
    if not DATE_PATTERN.match(value):
        raise ValueError(f"{field_name} must be in YYYY-MM-DD format")
    return value


def validate_nif(value) -> int:
    """Validate NIF as a non-negative integer."""
    try:
        nif = int(value) if value else 0
        if nif < 0:
            raise ValueError("NIF cannot be negative")
        return nif
    except (TypeError, ValueError):
        raise ValueError("NIF must be a valid integer")

async def get_employees() -> list[dict]:
    """
    Return a list of employees by calling an external web service.

    Returns:
        list[dict]: Response from the web service with columns:
        Name (str): Name of the employee (required)
        NIF (int): Tax identification number (defaults to 0)
        DateOfBirth (str): Date of birth in YYYY-MM-DD format (defaults to empty string)
        Address (str): Employee's address (defaults to empty string)
        CreatedOn (date time): Date and time when the employee was created (defaults to empty string)
        Phone (str): Employee's phone number (defaults to empty string)
    """
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(f"{BASE_URL}/GetEmployees")
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            raise Exception(f"Failed to fetch employees (HTTP {e.response.status_code})")
        except Exception:
            raise Exception("Failed to connect to employee service")

async def get_employee_by_id(employee_id: int) -> dict:
    """
    Return a specific employee by ID.

    Args:
        employee_id (int): The unique identifier of the employee

    Returns:
        dict: Employee data with fields:
        Id (int): Employee ID
        Name (str): Name of the employee
        NIF (int): Tax identification number
        DateOfBirth (str): Date of birth in YYYY-MM-DD format
        Address (str): Employee's address
        CreatedOn (datetime): Date and time when the employee was created
        Phone (str): Employee's phone number
    """
    validated_id = validate_employee_id(employee_id)
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(f"{BASE_URL}/GetEmployeeBy", params={"EmployeeId": validated_id})
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                raise Exception("Employee not found")
            raise Exception(f"Failed to fetch employee (HTTP {e.response.status_code})")
        except ValueError as e:
            raise Exception(str(e))
        except Exception:
            raise Exception("Failed to connect to employee service")

async def create_employee(name: str, nif: int = 0, date_of_birth: str = "", address: str = "") -> str:
    """
    Create a new employee.

    Args:
        name (str): Name of the employee (required)
        nif (int): Tax identification number (optional, defaults to 0)
        date_of_birth (str): Date of birth in YYYY-MM-DD format (optional)
        address (str): Employee's address (optional)

    Returns:
        str: Result message from the server
    """
    # Validate all inputs
    validated_name = validate_string(name, "Name", required=True)
    validated_nif = validate_nif(nif)
    validated_dob = validate_date(date_of_birth, "Date of birth")
    validated_address = validate_string(address, "Address", required=False)

    async with httpx.AsyncClient() as client:
        try:
            payload = {
                "EmployeeName": validated_name,
                "NIF": validated_nif,
                "DateOfBirth": validated_dob,
                "Address": validated_address
            }
            response = await client.post(f"{BASE_URL}/EmployeeCreate", json=payload)
            response.raise_for_status()
            return response.text
        except httpx.HTTPStatusError as e:
            raise Exception(f"Failed to create employee (HTTP {e.response.status_code})")
        except ValueError as e:
            raise Exception(str(e))
        except Exception:
            raise Exception("Failed to connect to employee service")

# MCP Protocol handler
async def handle_mcp_request(request_data: str) -> str:
    """Handle MCP protocol requests"""
    try:
        request = json.loads(request_data)
        
        if request.get("method") == "tools/list":
            # Return available tools
            return json.dumps({
                "jsonrpc": "2.0",
                "id": request.get("id"),
                "result": {
                    "tools": [
                        {
                            "name": "get_employees",
                            "description": "Get list of all employees",
                            "inputSchema": {
                                "type": "object",
                                "properties": {},
                                "required": []
                            }
                        },
                        {
                            "name": "get_employee_by_id",
                            "description": "Get a specific employee by their ID",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "employee_id": {
                                        "type": "integer",
                                        "description": "The unique identifier of the employee"
                                    }
                                },
                                "required": ["employee_id"]
                            }
                        },
                        {
                            "name": "create_employee",
                            "description": "Create a new employee record",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "name": {
                                        "type": "string",
                                        "description": "Name of the employee"
                                    },
                                    "nif": {
                                        "type": "integer",
                                        "description": "Tax identification number (optional)"
                                    },
                                    "date_of_birth": {
                                        "type": "string",
                                        "description": "Date of birth in YYYY-MM-DD format (optional)"
                                    },
                                    "address": {
                                        "type": "string",
                                        "description": "Employee's address (optional)"
                                    }
                                },
                                "required": ["name"]
                            }
                        }
                    ]
                }
            })
        
        elif request.get("method") == "tools/call":
            tool_name = request.get("params", {}).get("name")
            arguments = request.get("params", {}).get("arguments", {})

            if tool_name == "get_employees":
                result = await get_employees()
                return json.dumps({
                    "jsonrpc": "2.0",
                    "id": request.get("id"),
                    "result": {
                        "content": [{
                            "type": "text",
                            "text": json.dumps(result, indent=2)
                        }]
                    }
                })

            elif tool_name == "get_employee_by_id":
                employee_id = arguments.get("employee_id")
                if employee_id is None:
                    return json.dumps({
                        "jsonrpc": "2.0",
                        "id": request.get("id"),
                        "error": {
                            "code": -32602,
                            "message": "Missing required parameter: employee_id"
                        }
                    })
                try:
                    result = await get_employee_by_id(employee_id)
                    return json.dumps({
                        "jsonrpc": "2.0",
                        "id": request.get("id"),
                        "result": {
                            "content": [{
                                "type": "text",
                                "text": json.dumps(result, indent=2)
                            }]
                        }
                    })
                except Exception as e:
                    return json.dumps({
                        "jsonrpc": "2.0",
                        "id": request.get("id"),
                        "error": {
                            "code": -32602,
                            "message": str(e)
                        }
                    })

            elif tool_name == "create_employee":
                name = arguments.get("name")
                if not name:
                    return json.dumps({
                        "jsonrpc": "2.0",
                        "id": request.get("id"),
                        "error": {
                            "code": -32602,
                            "message": "Missing required parameter: name"
                        }
                    })
                try:
                    result = await create_employee(
                        name=name,
                        nif=arguments.get("nif", 0),
                        date_of_birth=arguments.get("date_of_birth", ""),
                        address=arguments.get("address", "")
                    )
                    return json.dumps({
                        "jsonrpc": "2.0",
                        "id": request.get("id"),
                        "result": {
                            "content": [{
                                "type": "text",
                                "text": result
                            }]
                        }
                    })
                except Exception as e:
                    return json.dumps({
                        "jsonrpc": "2.0",
                        "id": request.get("id"),
                        "error": {
                            "code": -32602,
                            "message": str(e)
                        }
                    })
        
        # Default error response
        return json.dumps({
            "jsonrpc": "2.0",
            "id": request.get("id"),
            "error": {
                "code": -32601,
                "message": "Method not found"
            }
        })
    
    except Exception as e:
        return json.dumps({
            "jsonrpc": "2.0",
            "id": request.get("id", None),
            "error": {
                "code": -32603,
                "message": f"Internal error: {str(e)}"
            }
        })

# Create interfaces
employees_interface = gr.Interface(
    fn=get_employees,
    inputs=None,
    outputs="json",
    title="Get Employees",
    description="Return the list of all employees"
)

employee_by_id_interface = gr.Interface(
    fn=get_employee_by_id,
    inputs=gr.Number(label="Employee ID", precision=0),
    outputs="json",
    title="Get Employee by ID",
    description="Return a specific employee by their ID"
)

create_employee_interface = gr.Interface(
    fn=create_employee,
    inputs=[
        gr.Textbox(label="Name", placeholder="Employee name (required)"),
        gr.Number(label="NIF", value=0, precision=0),
        gr.Textbox(label="Date of Birth", placeholder="YYYY-MM-DD"),
        gr.Textbox(label="Address", placeholder="Employee address")
    ],
    outputs="text",
    title="Create Employee",
    description="Create a new employee record"
)

mcp_interface = gr.Interface(
    fn=handle_mcp_request,
    inputs=gr.Textbox(label="MCP Request JSON"),
    outputs=gr.Textbox(label="MCP Response JSON"),
    title="MCP Server",
    description="MCP protocol endpoint"
)

# Combine interfaces
demo = gr.TabbedInterface(
    [employees_interface, employee_by_id_interface, create_employee_interface, mcp_interface],
    ["Get All Employees", "Get Employee by ID", "Create Employee", "MCP Protocol"]
)

if __name__ == "__main__":
    demo.launch(mcp_server=True, share=True)