2 Commits

6 changed files with 163 additions and 3 deletions
@@ -0,0 +1,64 @@
"""
Script para generar un servidor MCP a partir de código fuente recibido (por ejemplo, desde un LLM).
Crea un archivo Python en la carpeta de servidores MCP con el código proporcionado y un nombre único.
"""
import sys
import os
from datetime import datetime
from pathlib import Path
from fastmcp import FastMCP
SERVERS_DIR = Path(__file__).parent
mcp = FastMCP()
def generate_server(code: str, name: str = None) -> str:
if not name:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
name = f"server_llm_{timestamp}.py"
else:
if not name.endswith('.py'):
name += '.py'
server_path = SERVERS_DIR / name
with open(server_path, 'w', encoding='utf-8') as f:
f.write(code)
return str(server_path)
@mcp.tool(description="Genera un archivo de servidor MCP a partir de código fuente y un nombre opcional.")
def mcp_generate_server(code: str, name: str = None) -> str:
"""
Esta herramienta guarda el código fuente en un archivo Python con nombre opcional.
Args:
code: Código fuente del servidor MCP como string.
name: (opcional) Nombre del archivo. Si no se especifica, se generará uno con timestamp.
Requiere que el código incluya explícitamente: path="/"
Returns:
Ruta absoluta del archivo creado.
Ejemplo de uso mínimo:
mcp_generate_server(
code=\"\"\"from fastmcp import FastMCP
mcp = FastMCP()
@mcp.tool(description="Saluda al mundo.")
def hello():
return "Hola mundo"
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="127.0.0.1", port=4211, path="/")
\"\"\"
)
"""
# Validación: asegurar que el código incluya path="/"
if 'path="/"' not in code.replace(" ", "").replace("'", '"'):
raise ValueError('El código del servidor debe contener path="/".')
return generate_server(code, name)
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="127.0.0.1", port=4210, path="/")
# mcp.run(transport="stdio")
+1 -1
View File
@@ -129,5 +129,5 @@ def clear_folder(path: str) -> str:
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="127.0.0.1", port=4201, path="/fs")
mcp.run(transport="streamable-http", host="127.0.0.1", port=4201, path="/")
+1 -1
View File
@@ -87,6 +87,6 @@ def is_prime(n: int) -> bool:
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="127.0.0.1", port=4200, path="/math")
mcp.run(transport="streamable-http", host="127.0.0.1", port=4200, path="/")
# mcp.run(transport="stdio")
+1 -1
View File
@@ -65,5 +65,5 @@ def current_timestamp() -> float:
return datetime.datetime.now().timestamp()
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="127.0.0.1", port=4300, path="/tools")
mcp.run(transport="streamable-http", host="127.0.0.1", port=4300, path="/")
+34
View File
@@ -0,0 +1,34 @@
import os
import signal
import time
PID_FILE = "mcps.pid"
def cerrar_procesos(pid_file):
if not os.path.exists(pid_file):
print(f"❌ No se encontró el archivo {pid_file}")
return
with open(pid_file, "r") as f:
pids = [line.strip() for line in f if line.strip().isdigit()]
for pid_str in pids:
pid = int(pid_str)
try:
os.kill(pid, signal.SIGTERM)
print(f"✔ Proceso {pid} terminado con SIGTERM.")
except ProcessLookupError:
print(f"⚠ El proceso {pid} no existe o ya fue terminado.")
except PermissionError:
print(f"❌ No tienes permiso para terminar el proceso {pid}.")
except Exception as e:
print(f"❌ Error al cerrar el proceso {pid}: {e}")
try:
os.remove(pid_file)
print(f"🧹 Archivo {pid_file} eliminado.")
except Exception as e:
print(f"⚠ No se pudo eliminar {pid_file}: {e}")
if __name__ == "__main__":
cerrar_procesos(PID_FILE)
+62
View File
@@ -0,0 +1,62 @@
import os
import subprocess
import sys
import re
from pathlib import Path
ROOT = Path(__file__).resolve().parent
PYTHON_PATH = (ROOT.parent / ROOT.parent / ".venv" / "Scripts" / "python.exe").resolve()
PID_FILE = ROOT / "mcps.pid"
# Carpeta de MCPs
MCPS_DIR = (ROOT / "../domains/Llms/MCPs/McpServers").resolve()
print("🚀 Iniciando todos los MCPs en el directorio:", MCPS_DIR)
# Verificar intérprete Python
if not PYTHON_PATH.exists():
print(f"❌ No se encontró Python en: {PYTHON_PATH}")
sys.exit(1)
# Limpiar archivo de PIDs previo
if PID_FILE.exists():
PID_FILE.unlink()
# Expresión regular para capturar la URL de uvicorn
URL_REGEX = re.compile(r"Uvicorn running on (?P<url>http://[^\s]+)")
# Ejecutar todos los archivos .py en MCPS_DIR
for script_path in sorted(MCPS_DIR.glob("*.py")):
if script_path.name.startswith("__"):
continue # Evita archivos como __init__.py
try:
proc = subprocess.Popen(
[str(PYTHON_PATH), str(script_path)],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True
)
# Leer la salida para encontrar la URL
url = None
for line in proc.stdout:
match = URL_REGEX.search(line)
if match:
url = match.group("url")
break
# Guardar el PID
with open(PID_FILE, "a", encoding="utf-8") as f:
f.write(str(proc.pid) + "\n")
print(f"{script_path.relative_to(ROOT)} lanzado con PID {proc.pid}")
if url:
print(f" 🛰️ Servidor accesible en: {url}")
else:
print(" ⚠ No se detectó URL del servidor.")
except Exception as e:
print(f"❌ Error al lanzar {script_path.name}:\n{e}")
print("✅ Todos los scripts han sido procesados. PIDs guardados en mcps.pid.")