En este tutorial, implementaremos un cliente del Protocolo de contexto del modelo personalizado (MCP) que usa Gemini. Al final de este tutorial, podrá conectar sus propias aplicaciones de IA con servidores MCP, desbloqueando nuevas capacidades poderosas para sobrealimentar sus proyectos.
API GEMINI
Usaremos el modelo Flash Gemini 2.0 para este tutorial.
Para obtener su llave de la API de Géminis, visite Llave de API de Géminis de Google página y siga las instrucciones.
Una vez que tenga la llave, guárdela de manera segura, la necesitará más tarde.
Nodo.js
Algunos de los servidores MCP requieren Node.js para ejecutarse. Descarga lo último versión de node.js de nodejs.org
- Ejecute el instalador.
- Deje todas las configuraciones de manera predeterminada y complete la instalación.
API de Servicios del Parque Nacional
Para este tutorial, expondremos el servidor MCP de los Servicios de Parques Nacionales a nuestro cliente. Para usar la API del Servicio de Parques Nacionales, puede solicitar una clave API visitando este enlace y llenando un formulario corto. Una vez enviado, la clave API se enviará a su correo electrónico.
Asegúrese de mantener esta clave accesible: la usaremos en breve.
Instalación de bibliotecas de Python
En el símbolo del sistema, ingrese el siguiente código para instalar las bibliotecas de Python:
pip install mcp python-dotenv google-genaiCreación del archivo MCP.JSON
A continuación, cree un archivo llamado mcp.json.
Este archivo almacenará detalles de configuración sobre los servidores MCP a los que su cliente se conectará.
Una vez que se cree el archivo, agregue el siguiente contenido inicial:
{
"mcpServers": {
"nationalparks": {
"command": "npx",
"args": ["-y", "mcp-server-nationalparks"],
"env": {
"NPS_API_KEY": <”YOUR_NPS_API_KEY”>
}
}
}
}Reemplace
Creación del archivo .env
Cree un archivo .env en el mismo directorio que el archivo mcp.json e ingrese el siguiente código:
GEMINI_API_KEY = <YOUR_GEMINI_API_KEY>Reemplace
Ahora crearemos un Client.py Archivo para implementar nuestro cliente MCP. Asegúrese de que este archivo esté en el mismo directorio que mcp.json y .env
Estructura de cliente básica
Primero importaremos las bibliotecas necesarias y crearemos una clase de cliente básica
import asyncio
import json
import os
from typing import List, Optional
from contextlib import AsyncExitStack
import warnings
from google import genai
from google.genai import types
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from dotenv import load_dotenv
load_dotenv()
warnings.filterwarnings("ignore", category=ResourceWarning)
def clean_schema(schema): # Cleans the schema by keeping only allowed keys
allowed_keys = {"type", "properties", "required", "description", "title", "default", "enum"}
return {k: v for k, v in schema.items() if k in allowed_keys}
class MCPGeminiAgent:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.genai_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
self.model = "gemini-2.0-flash"
self.tools = None
self.server_params = None
self.server_name = NoneEl método __init__ inicializa el McPgeminiagent configurando un administrador de sesión asíncrono, cargando el cliente de la API de Gemini y preparando a los marcadores de posición para la configuración del modelo, las herramientas y los detalles del servidor.
Se sienta las bases para administrar las conexiones del servidor e interactuar con el modelo Gemini.
Seleccionando el servidor MCP
async def select_server(self):
with open('mcp.json', 'r') as f:
mcp_config = json.load(f)
servers = mcp_config['mcpServers']
server_names = list(servers.keys())
print("Available MCP servers:")
for idx, name in enumerate(server_names):
print(f" {idx+1}. {name}")
while True:
try:
choice = int(input(f"Please select a server by number [1-{len(server_names)}]: "))
if 1 <= choice <= len(server_names):
break
else:
print("That number is not valid. Please try again.")
except ValueError:
print("Please enter a valid number.")
self.server_name = server_names[choice-1]
server_cfg = servers[self.server_name]
command = server_cfg['command']
args = server_cfg.get('args', [])
env = server_cfg.get('env', None)
self.server_params = StdioServerParameters(
command=command,
args=args,
env=env
)Este método solicita al usuario que elija un servidor de las opciones disponibles enumeradas en MCP.JSON. Carga y prepara los parámetros de conexión del servidor seleccionado para su uso posterior.
Conectarse al servidor MCP
async def connect(self):
await self.select_server()
self.stdio_transport = await self.exit_stack.enter_async_context(stdio_client(self.server_params))
self.stdio, self.write = self.stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
print(f"Successfully connected to: {self.server_name}")
# List available tools for this server
mcp_tools = await self.session.list_tools()
print("\nAvailable MCP tools for this server:")
for tool in mcp_tools.tools:
print(f"- {tool.name}: {tool.description}")Esto establece una conexión asincrónica con el servidor MCP seleccionado utilizando el transporte STDIO. Inicializa la sesión MCP y recupera las herramientas disponibles del servidor.
Manejo de la consulta del usuario y las llamadas de herramientas
async def agent_loop(self, prompt: str) -> str:
contents = [types.Content(role="user", parts=[types.Part(text=prompt)])]
mcp_tools = await self.session.list_tools()
tools = types.Tool(function_declarations=[
{
"name": tool.name,
"description": tool.description,
"parameters": clean_schema(getattr(tool, "inputSchema", {}))
}
for tool in mcp_tools.tools
])
self.tools = tools
response = await self.genai_client.aio.models.generate_content(
model=self.model,
contents=contents,
config=types.GenerateContentConfig(
temperature=0,
tools=[tools],
),
)
contents.append(response.candidates[0].content)
turn_count = 0
max_tool_turns = 5
while response.function_calls and turn_count < max_tool_turns:
turn_count += 1
tool_response_parts: List[types.Part] = []
for fc_part in response.function_calls:
tool_name = fc_part.name
args = fc_part.args or {}
print(f"Invoking MCP tool '{tool_name}' with arguments: {args}")
tool_response: dict
try:
tool_result = await self.session.call_tool(tool_name, args)
print(f"Tool '{tool_name}' executed.")
if tool_result.isError:
tool_response = {"error": tool_result.content[0].text}
else:
tool_response = {"result": tool_result.content[0].text}
except Exception as e:
tool_response = {"error": f"Tool execution failed: {type(e).__name__}: {e}"}
tool_response_parts.append(
types.Part.from_function_response(
name=tool_name, response=tool_response
)
)
contents.append(types.Content(role="user", parts=tool_response_parts))
print(f"Added {len(tool_response_parts)} tool response(s) to the conversation.")
print("Requesting updated response from Gemini...")
response = await self.genai_client.aio.models.generate_content(
model=self.model,
contents=contents,
config=types.GenerateContentConfig(
temperature=1.0,
tools=[tools],
),
)
contents.append(response.candidates[0].content)
if turn_count >= max_tool_turns and response.function_calls:
print(f"Stopped after {max_tool_turns} tool calls to avoid infinite loops.")
print("All tool calls complete. Displaying Gemini's final response.")
return responseEste método envía el indicador del usuario a Gemini, procesa cualquier llamada de herramienta devuelta por el modelo, ejecuta las herramientas MCP correspondientes y refina iterativamente la respuesta. Administra interacciones múltiples entre Gemini y las herramientas del servidor.
Ciclo de chat interactivo
async def chat(self):
print(f"\nMCP-Gemini Assistant is ready and connected to: {self.server_name}")
print("Enter your question below, or type 'quit' to exit.")
while True:
try:
query = input("\nYour query: ").strip()
if query.lower() == 'quit':
print("Session ended. Goodbye!")
break
print(f"Processing your request...")
res = await self.agent_loop(query)
print("\nGemini's answer:")
print(res.text)
except KeyboardInterrupt:
print("\nSession interrupted. Goodbye!")
break
except Exception as e:
print(f"\nAn error occurred: {str(e)}")Esto proporciona una interfaz de línea de comandos donde los usuarios pueden enviar consultas y recibir respuestas de Gemini, continuamente hasta que salgan de la sesión.
Limpiar recursos
async def cleanup(self):
await self.exit_stack.aclose()Esto cierra el contexto asincrónico y limpia todos los recursos abiertos como la sesión y la pila de conexión con gracia.
Principal punto de entrada
async def main():
agent = MCPGeminiAgent()
try:
await agent.connect()
await agent.chat()
finally:
await agent.cleanup()
if __name__ == "__main__":
import sys
import os
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Session interrupted. Goodbye!")
finally:
sys.stderr = open(os.devnull, "w")Esta es la lógica de ejecución principal.
Aparte de Main (), todos los demás métodos son parte de Mcpgeminiagente clase. Puede encontrar el archivo completo client.py aquí.
Ejecute el siguiente mensaje en el terminal para ejecutar su cliente:
El cliente lo hará:
- Lea el archivo MCP.JSON para enumerar los diferentes servidores MCP disponibles.
- Solicite al usuario que seleccione uno de los servidores enumerados.
- Conéctese al servidor MCP seleccionado utilizando la configuración de configuración y el entorno proporcionadas.
- Interactuar con el modelo Géminis a través de una serie de consultas y respuestas.
- Permitir a los usuarios emitir indicaciones, ejecutar herramientas y procesar respuestas de iteración con el modelo.
- Proporcione una interfaz de línea de comandos para que los usuarios se involucren con el sistema y reciban resultados en tiempo real.
- Asegure la limpieza adecuada de los recursos después de que finalice la sesión, cierre las conexiones y la liberación de la memoria.
Soy un graduado de ingeniería civil (2022) de Jamia Millia Islamia, Nueva Delhi, y tengo un gran interés en la ciencia de datos, especialmente las redes neuronales y su aplicación en varias áreas.
