Servidor MCP
Lengua china
Descripción general del proyecto
Basado en FastAPI y MCP (Protocolo de Contexto de Modelo), este proyecto permite la interacción contextual estandarizada entre modelos de IA y entornos de desarrollo. Mejora la escalabilidad y el mantenimiento de las aplicaciones de IA al simplificar la implementación de modelos, proporcionar endpoints API eficientes y garantizar la consistencia en la entrada y salida de los modelos, facilitando a los desarrolladores la integración y gestión de tareas de IA.
MCP (Protocolo de Contexto de Modelo) es un protocolo unificado para la interacción contextual entre modelos de IA y entornos de desarrollo. Este proyecto proporciona una implementación de servidor MCP basada en Python que admite funciones básicas del protocolo MCP, como la inicialización, el muestreo y la gestión de sesiones.
Características
- JSON-RPC 2.0 : Comunicación de solicitud-respuesta basada en el protocolo estándar JSON-RPC 2.0
- Conexión SSE : Compatibilidad con conexiones de eventos enviados por el servidor para notificaciones en tiempo real
- Diseño modular : arquitectura modular para una fácil ampliación y personalización
- Procesamiento asincrónico : servicio de alto rendimiento que utiliza FastAPI y E/S asincrónica
- Cliente completo : incluye una implementación de cliente de prueba completa
Estructura del proyecto
mcp_server/
├── mcp_server.py # MCP server main program
├── mcp_client.py # MCP client test program
├── routers/
│ ├── __init__.py # Router package initialization
│ └── base_router.py # Base router implementation
├── requirements.txt # Project dependencies
└── README.md # Project documentation
Instalación
- Clonar el repositorio:
git clone https://github.com/freedanfan/mcp_server.git
cd mcp_server
- Instalar dependencias:
pip install -r requirements.txt
Uso
Iniciando el servidor
De forma predeterminada, el servidor se iniciará en 127.0.0.1:12000 . Puede personalizar el host y el puerto mediante variables de entorno:
export MCP_SERVER_HOST=0.0.0.0
export MCP_SERVER_PORT=8000
python mcp_server.py
Ejecución del cliente
Ejecute el cliente en otra terminal:
Si el servidor no se ejecuta en la dirección predeterminada, puede configurar una variable de entorno:
export MCP_SERVER_URL="http://your-server-address:port"
python mcp_client.py
Puntos finales de API
El servidor proporciona los siguientes puntos finales de API:
- Ruta raíz (
/ ): proporciona información del servidor - Punto final de API (
/api ): maneja solicitudes JSON-RPC - Punto final SSE (
/sse ): maneja conexiones SSE
Implementación del protocolo MCP
Flujo de inicialización
- El cliente se conecta al servidor a través de SSE
- El servidor devuelve la URI del punto final de la API
- El cliente envía una solicitud de inicialización con la versión del protocolo y las capacidades
- El servidor responde a la solicitud de inicialización, devolviendo las capacidades del servidor.
Solicitud de muestreo
Los clientes pueden enviar solicitudes de muestreo con indicaciones:
{
"jsonrpc": "2.0",
"id": "request-id",
"method": "sample",
"params": {
"prompt": "Hello, please introduce yourself."
}
}
El servidor devolverá los resultados del muestreo:
{
"jsonrpc": "2.0",
"id": "request-id",
"result": {
"content": "This is a response to the prompt...",
"usage": {
"prompt_tokens": 10,
"completion_tokens": 50,
"total_tokens": 60
}
}
}
Cerrar una sesión
Los clientes pueden enviar una solicitud de apagado:
{
"jsonrpc": "2.0",
"id": "request-id",
"method": "shutdown",
"params": {}
}
El servidor se apagará correctamente:
{
"jsonrpc": "2.0",
"id": "request-id",
"result": {
"status": "shutting_down"
}
}
Extensiones de desarrollo
Agregar nuevos métodos
Para agregar nuevos métodos MCP, agregue una función de controlador a la clase MCPServer y regístrela en el método _register_methods :
def handle_new_method(self, params: dict) -> dict:
"""Handle new method"""
logger.info(f"Received new method request: {params}")
# Processing logic
return {"result": "success"}
def _register_methods(self):
# Register existing methods
self.router.register_method("initialize", self.handle_initialize)
self.router.register_method("sample", self.handle_sample)
self.router.register_method("shutdown", self.handle_shutdown)
# Register new method
self.router.register_method("new_method", self.handle_new_method)
Integración de modelos de IA
Para integrar modelos de IA reales, modifique el método handle_sample :
async def handle_sample(self, params: dict) -> dict:
"""Handle sampling request"""
logger.info(f"Received sampling request: {params}")
# Get prompt
prompt = params.get("prompt", "")
# Call AI model API
# For example: using OpenAI API
response = await openai.ChatCompletion.acreate(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
content = response.choices[0].message.content
usage = response.usage
return {
"content": content,
"usage": {
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens
}
}
Solución de problemas
Problemas comunes
- Errores de conexión : asegúrese de que el servidor esté en ejecución y que el cliente esté usando la URL del servidor correcta
- 405 Método no permitido : asegúrese de que el cliente esté enviando solicitudes al punto final de API correcto
- Error de conexión SSE : Verifique las conexiones de red y la configuración del firewall
Explotación florestal
Tanto el servidor como el cliente proporcionan registros detallados. Consulte los registros para obtener más información:
# Increase log level
export PYTHONPATH=.
python -m logging -v DEBUG -m mcp_server
Referencias
Licencia
Este proyecto está licenciado bajo la Licencia MIT. Consulte el archivo de LICENCIA para más detalles.