from mcp.server.fastmcp import FastMCP
import httpx
from typing import List, Optional, Dict, Any
import os
import json
# Inicializar servidor MCP
mcp = FastMCP("MelviChat")
BASE_URL = "http://127.0.0.1:8000"
TOKEN_FILE = ".melvi_token"
def save_token(token: str):
"""Guardar token en archivo local para persistencia."""
with open(TOKEN_FILE, "w") as f:
f.write(token)
def load_token() -> Optional[str]:
"""Cargar token desde archivo local."""
if os.path.exists(TOKEN_FILE):
with open(TOKEN_FILE, "r") as f:
return f.read().strip()
return None
@mcp.tool()
async def login(email: str, password: str) -> Dict[str, Any]:
"""
Iniciar sesión en MelviChat.
Guarda el token localmente para no tener que enviarlo en cada petición.
"""
async with httpx.AsyncClient() as client:
try:
resp = await client.post(f"{BASE_URL}/tenants/login", json={"email": email, "password": password})
resp.raise_for_status()
data = resp.json()
token = data.get("access_token")
if token:
save_token(token)
return {"status": "success", "message": "Login exitoso. Token guardado.", "token_preview": token[:10] + "..."}
return data
except httpx.HTTPStatusError as e:
return {"error": f"Login fallido: {e.response.text}", "status": e.response.status_code}
except Exception as e:
return {"error": str(e)}
@mcp.tool()
async def chat_with_database(prompt: str) -> Dict[str, Any]:
"""
Chatear con tu Base de Datos.
Envía una pregunta en lenguaje natural y MelviChat (tu backend) la interpretará,
generará SQL, ejecutará la consulta y devolverá los resultados.
Ejemplos:
- "¿Cuántos usuarios se registraron hoy?"
- "Muestrame las ventas del último mes"
"""
token = load_token()
if not token:
return {"error": "No has iniciado sesión. Usa la herramienta 'login' primero."}
async with httpx.AsyncClient(timeout=60.0) as client:
try:
resp = await client.post(
f"{BASE_URL}/ai/query",
headers={"Authorization": f"Bearer {token}"},
json={"prompt": prompt, "history": []}
)
resp.raise_for_status()
return resp.json()
except httpx.HTTPStatusError as e:
return {"error": f"Error en IA: {e.response.text}"}
except Exception as e:
return {"error": str(e)}
@mcp.tool()
async def get_database_structure() -> Dict[str, Any]:
"""
Obtener la estructura (esquema) de la base de datos.
Útil para saber qué tablas y campos existen antes de preguntar.
"""
token = load_token()
# Nota: El endpoint /db/schema parece no requerir auth en el código actual (dynamic_api.py),
# pero es buena práctica enviar el token si se requiere en el futuro o middleware.
# Si el endpoint es público en backend, esto funcionará. Si requiere auth, se añade header.
# Revisando dynamic_api.py, no tiene `Depends(verify_tenant_token)` explícito en el router level
# pero `get_table_data` usa `Depends(get_db)`.
# Probemos sin token, si falla, añadimos auth.
async with httpx.AsyncClient() as client:
try:
# Intentar obtener esquema completo
resp = await client.get(f"{BASE_URL}/db/schema")
if resp.status_code == 200:
return resp.json()
# Si falla, intentar listar tablas
resp = await client.get(f"{BASE_URL}/db/tables")
return resp.json()
except Exception as e:
return {"error": str(e)}
@mcp.resource("melvi://profile")
async def get_my_profile() -> str:
"""
Recurso que devuelve el perfil del usuario actual en formato JSON.
"""
token = load_token()
if not token:
return "Error: No logged in"
async with httpx.AsyncClient() as client:
try:
resp = await client.get(
f"{BASE_URL}/tenants/profile",
headers={"Authorization": f"Bearer {token}"}
)
return json.dumps(resp.json(), indent=2)
except Exception as e:
return f"Error: {str(e)}"
if __name__ == "__main__":
mcp.run()
