Construyendo un chatbot MCP сustom | Hacia la ciencia de los datos

Un método para estandarizar la comunicación entre aplicaciones de IA y herramientas externas o fuentes de datos. Esta estandarización ayuda a reducir el número de integraciones necesarias (De n*m ​​a n+m)

• Puede usar servidores MCP construidos en la comunidad cuando necesita una funcionalidad común, ahorrar tiempo y evitar la necesidad de reinventar la rueda cada vez.
• También puede exponer sus propias herramientas y recursos, haciéndolos disponibles para que otros lo usen.

En Mi artículo anteriorconstruimos la caja de herramientas Analytics (una colección de herramientas que podrían automatizar su rutina diaria). Construimos un servidor MCP y utilizamos sus capacidades con clientes existentes como MCP Inspector o Claude Desktop.

Ahora, queremos usar esas herramientas directamente en nuestras aplicaciones AI. Para hacer eso, construamos nuestro propio cliente MCP. Escribiremos un código de nivel bastante bajo, que también le dará una imagen más clara de cómo las herramientas como Claude Code interactúan con MCP debajo del capó.

Además, me gustaría implementar la función que está actualmente (Julio de 2025) Falta en Claude Desktop: la capacidad del LLM para verificar automáticamente si tiene una plantilla de inmediato adecuada para la tarea en cuestión y usarla. En este momento, debe elegir la plantilla manualmente, lo cual no es muy conveniente.

Como beneficio adicional, también compartiré una implementación de alto nivel utilizando el marco Smolagents, que es ideal para escenarios en los que trabaja solo con herramientas MCP y no necesita mucha personalización.

Descripción general del protocolo MCP

Aquí hay un resumen rápido del MCP para garantizar que estemos en la misma página. MCP es un protocolo desarrollado por Anthrope para estandarizar la forma en que los LLM interactúan con el mundo exterior.

Sigue una arquitectura de cliente cliente y consta de tres componentes principales:

• Anfitrión es la aplicación orientada al usuario.
• Cliente de MCP es un componente dentro del host que establece una conexión uno a uno con el servidor y se comunica utilizando mensajes definidos por el protocolo MCP.
• Servidor MCP Expone capacidades tales como plantillas, recursos y herramientas de inmediato.

Ya que ya lo hemos implementó el servidor MCP Antes, esta vez nos centraremos en construir el cliente MCP. Comenzaremos con una implementación relativamente simple y luego agregaremos la capacidad de seleccionar dinámicamente plantillas de solicitud en la marcha.

Puede encontrar el código completo en Github.

Construyendo el chatbot MCP

Comencemos con la configuración inicial: cargaremos la tecla API antrópica desde un archivo de configuración y ajustaremos a Python’s asyncio Bucle de eventos para apoyar bucles de eventos anidados.

# Load configuration and environment
with open('../../config.json') as f:
    config = json.load(f)
os.environ["ANTHROPIC_API_KEY"] = config['ANTHROPIC_API_KEY']

nest_asyncio.apply()

Comencemos por construir un esqueleto de nuestro programa para obtener una imagen clara de la arquitectura de alto nivel de la aplicación.

async def main():
    """Main entry point for the MCP ChatBot application."""
    chatbot = MCP_ChatBot()
    try:
        await chatbot.connect_to_servers()
        await chatbot.chat_loop()
    finally:
        await chatbot.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

Comenzamos creando una instancia del MCP_ChatBot clase. El chatbot comienza descubriendo las capacidades de MCP disponibles (iterando a través de todos los servidores MCP configurados, estableciendo conexiones y solicitando sus listas de capacidades).

Una vez que se configuren las conexiones, inicializaremos un bucle infinito donde el chatbot escuche las consultas del usuario, llame a las herramientas cuando sea necesario y continúa este ciclo hasta que el proceso se detenga manualmente.

Finalmente, realizaremos un paso de limpieza para cerrar todas las conexiones abiertas.

Ahora caminemos por cada etapa con más detalle.

Inicializar la clase de chatbot

Comencemos creando la clase y definiendo el __init__ método. Los campos principales de la clase de chatbot son:

• exit_stack administra el ciclo de vida de múltiples hilos de async (conexiones a los servidores MCP), asegurando que todas las conexiones se cerren adecuadamente, incluso si enfrentamos un error durante la ejecución. Esta lógica se implementa en el cleanup función.
• anthropic es un cliente para la API antrópica que se usa para enviar mensajes a LLM.
• available_tools y available_prompts son las listas de herramientas y indicaciones expuestas por todos los servidores MCP a los que estamos conectados.
• sessions es un mapeo de herramientas, indicaciones y recursos a sus respectivas sesiones de MCP. Esto permite que el chatbot enrique las solicitudes al servidor MCP correcto cuando el LLM selecciona una herramienta específica.

class MCP_ChatBot:
  """
  MCP (Model Context Protocol) ChatBot that connects to multiple MCP servers
  and provides a conversational interface using Anthropic's Claude.
    
  Supports tools, prompts, and resources from connected MCP servers.
  """
    
  def __init__(self):
    self.exit_stack = AsyncExitStack() 
    self.anthropic = Anthropic() # Client for Anthropic API
    self.available_tools = [] # Tools from all connected servers
    self.available_prompts = [] # Prompts from all connected servers  
    self.sessions = {} # Maps tool/prompt/resource names to MCP sessions

  async def cleanup(self):
    """Clean up resources and close all connections."""
    await self.exit_stack.aclose()

Conectarse a servidores

La primera tarea para nuestro chatbot es iniciar conexiones con todos los servidores MCP configurados y descubrir qué capacidades podemos usar.

La lista de servidores MCP a los que nuestro agente puede conectarse se define en el server_config.json archivo. He configurado conexiones con tres servidores MCP:

• analista_toolkit es mi implementación de las herramientas analíticas cotidianas que discutimos en el artículo anterior,
• Sistema de archivos permite que el agente trabaje con archivos,
• Buscar Ayuda a LLMS a recuperar el contenido de las páginas web y a convertirlo de HTML a Markdown para una mejor legibilidad.

{
  "mcpServers": {
    "analyst_toolkit": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/github/mcp-analyst-toolkit/src/mcp_server",
        "run",
        "server.py"
      ],
      "env": {
          "GITHUB_TOKEN": "your_github_token"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/marie/Desktop",
        "/Users/marie/Documents/github"
      ]
    },
    "fetch": {
        "command": "uvx",
        "args": ["mcp-server-fetch"]
      }
  }
}

Primero, leeremos el archivo de configuración, lo analizaremos y luego nos conectaremos a cada servidor listado.

async def connect_to_servers(self):
  """Load server configuration and connect to all configured MCP servers."""
  try:
    with open("server_config.json", "r") as file:
      data = json.load(file)
    
    servers = data.get("mcpServers", {})
    for server_name, server_config in servers.items():
      await self.connect_to_server(server_name, server_config)
  except Exception as e:
    print(f"Error loading server config: {e}")
    traceback.print_exc()
    raise

Para cada servidor, realizamos varios pasos para establecer la conexión:

• A nivel de transporte, nosotros Inicie el servidor MCP como un proceso STDIO y obtenga transmisiones para enviar y recibir mensajes.
• A nivel de sesióncreamos un ClientSession incorporando las transmisiones, y luego realizamos el apretón de manos de MCP llamando initialize método.
• Registramos los objetos de la sesión y el transporte en el gerente de contexto exit_stack Para garantizar que todas las conexiones se cierren correctamente al final.
• El último paso es Registrar capacidades del servidor. Envolvemos esta funcionalidad en una función separada, y la discutiremos en breve.

async def connect_to_server(self, server_name, server_config):
    """Connect to a single MCP server and register its capabilities."""
    try:
      server_params = StdioServerParameters(**server_config)
      stdio_transport = await self.exit_stack.enter_async_context(
          stdio_client(server_params)
      )
      read, write = stdio_transport
      session = await self.exit_stack.enter_async_context(
          ClientSession(read, write)
      )
      await session.initialize()
      await self._register_server_capabilities(session, server_name)
            
    except Exception as e:
      print(f"Error connecting to {server_name}: {e}")
      traceback.print_exc()

El registro de las capacidades implica iterar sobre todas las herramientas, indicaciones y recursos recuperados de la sesión. Como resultado, actualizamos las variables internas sessions (Mapeo entre recursos y una sesión particular entre el cliente MCP y el servidor), available_prompts y available_tools.

async def _register_server_capabilities(self, session, server_name):
  """Register tools, prompts and resources from a single server."""
  capabilities = [
    ("tools", session.list_tools, self._register_tools),
    ("prompts", session.list_prompts, self._register_prompts), 
    ("resources", session.list_resources, self._register_resources)
  ]
  
  for capability_name, list_method, register_method in capabilities:
    try:
      response = await list_method()
      await register_method(response, session)
    except Exception as e:
      print(f"Server {server_name} doesn't support {capability_name}: {e}")

async def _register_tools(self, response, session):
  """Register tools from server response."""
  for tool in response.tools:
    self.sessions[tool.name] = session
    self.available_tools.append({
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    })

async def _register_prompts(self, response, session):
  """Register prompts from server response."""
  if response and response.prompts:
    for prompt in response.prompts:
        self.sessions[prompt.name] = session
        self.available_prompts.append({
            "name": prompt.name,
            "description": prompt.description,
            "arguments": prompt.arguments
        })

async def _register_resources(self, response, session):
  """Register resources from server response."""
  if response and response.resources:
    for resource in response.resources:
        resource_uri = str(resource.uri)
        self.sessions[resource_uri] = session

Al final de esta etapa, nuestro MCP_ChatBot El objeto tiene todo lo que necesita para comenzar a interactuar con los usuarios:

• Se establecen las conexiones a todos los servidores MCP configurados,
• Se registran todas las indicaciones, recursos y herramientas, incluidas las descripciones necesarias para que LLM comprenda cómo usar estas capacidades,
• Se almacenan mapeos entre estos recursos y sus respectivas sesiones, por lo que sabemos exactamente dónde enviar cada solicitud.

Bucle de chat

Entonces, es hora de comenzar nuestra chat con los usuarios creando el chat_loop función.

Primero compartiremos todos los comandos disponibles con el usuario:

• Listado de recursos, herramientas y indicaciones
• Ejecución de una llamada de herramienta
• Ver un recurso
• Usando una plantilla de inmediato
• renunciando al chat (Es importante tener una forma clara de salir del bucle infinito).

Después de eso, ingresaremos un bucle infinito donde, en función de la entrada del usuario, ejecutaremos la acción apropiada: si se trata de uno de los comandos anteriores o haciendo una solicitud al LLM.

async def chat_loop(self):
  """Main interactive chat loop with command processing."""
  print("\nMCP Chatbot Started!")
  print("Commands:")
  print("  quit                           - Exit the chatbot")
  print("  @periods                       - Show available changelog periods") 
  print("  @<period>                      - View changelog for specific period")
  print("  /tools                         - List available tools")
  print("  /tool <name> <arg1=value1>     - Execute a tool with arguments")
  print("  /prompts                       - List available prompts")
  print("  /prompt <name> <arg1=value1>   - Execute a prompt with arguments")
  
  while True:
    try:
      query = input("\nQuery: ").strip()
      if not query:
          continue

      if query.lower() == 'quit':
          break
      
      # Handle resource requests (@command)
      if query.startswith('@'):
        period = query[1:]
        resource_uri = "changelog://periods" if period == "periods" else f"changelog://{period}"
        await self.get_resource(resource_uri)
        continue
      
      # Handle slash commands
      if query.startswith('/'):
        parts = self._parse_command_arguments(query)
        if not parts:
          continue
            
        command = parts[0].lower()
        
        if command == '/tools':
          await self.list_tools()
        elif command == '/tool':
          if len(parts) < 2:
            print("Usage: /tool <name> <arg1=value1> <arg2=value2>")
            continue
            
          tool_name = parts[1]
          args = self._parse_prompt_arguments(parts[2:])
          await self.execute_tool(tool_name, args)
        elif command == '/prompts':
          await self.list_prompts()
        elif command == '/prompt':
          if len(parts) < 2:
            print("Usage: /prompt <name> <arg1=value1> <arg2=value2>")
            continue
          
          prompt_name = parts[1]
          args = self._parse_prompt_arguments(parts[2:])
          await self.execute_prompt(prompt_name, args)
        else:
          print(f"Unknown command: {command}")
        continue
      
      # Process regular queries
      await self.process_query(query)
            
    except Exception as e:
      print(f"\nError in chat loop: {e}")
      traceback.print_exc()

Hay un montón de funciones auxiliares para analizar argumentos y devolver las listas de herramientas y indicaciones disponibles que registramos anteriormente. Como es bastante sencillo, no entraré en muchos detalles aquí. Puedes comprobar el código completo Si estás interesado.

En su lugar, profundicemos en cómo funcionan las interacciones entre el cliente MCP y el servidor en diferentes escenarios.

Al trabajar con recursos, usamos el self.sessions Mapeo para encontrar la sesión apropiada (con una opción de respaldo si es necesario) y luego use esa sesión para leer el recurso.

async def get_resource(self, resource_uri):
  """Retrieve and display content from an MCP resource."""
  session = self.sessions.get(resource_uri)
  
  # Fallback: find any session that handles this resource type
  if not session and resource_uri.startswith("changelog://"):
    session = next(
        (sess for uri, sess in self.sessions.items() 
         if uri.startswith("changelog://")), 
        None
    )
      
  if not session:
    print(f"Resource '{resource_uri}' not found.")
    return

  try:
    result = await session.read_resource(uri=resource_uri)
    if result and result.contents:
        print(f"\nResource: {resource_uri}")
        print("Content:")
        print(result.contents[0].text)
    else:
        print("No content available.")
  except Exception as e:
    print(f"Error reading resource: {e}")
    traceback.print_exc()

Para ejecutar una herramienta, seguimos un proceso similar: comience por encontrar la sesión y luego usarla para llamar a la herramienta, pasando su nombre y argumentos.

async def execute_tool(self, tool_name, args):
  """Execute an MCP tool directly with given arguments."""
  session = self.sessions.get(tool_name)
  if not session:
      print(f"Tool '{tool_name}' not found.")
      return
  
  try:
      result = await session.call_tool(tool_name, arguments=args)
      print(f"\nTool '{tool_name}' result:")
      print(result.content)
  except Exception as e:
      print(f"Error executing tool: {e}")
      traceback.print_exc()

No es una sorpresa aquí. El mismo enfoque funciona para ejecutar el aviso.

async def execute_prompt(self, prompt_name, args):
    """Execute an MCP prompt with given arguments and process the result."""
    session = self.sessions.get(prompt_name)
    if not session:
        print(f"Prompt '{prompt_name}' not found.")
        return
    
    try:
        result = await session.get_prompt(prompt_name, arguments=args)
        if result and result.messages:
            prompt_content = result.messages[0].content
            text = self._extract_prompt_text(prompt_content)
            
            print(f"\nExecuting prompt '{prompt_name}'...")
            await self.process_query(text)
    except Exception as e:
        print(f"Error executing prompt: {e}")
        traceback.print_exc()

El único caso de uso importante que aún no hemos cubierto es manejar una entrada general de forma libre de un usuario (no uno de comandos específicos).
En este caso, primero enviamos la solicitud inicial al LLM, luego analizamos la salida, definiendo si hay alguna llamada de herramienta. Si las llamadas de herramientas están presentes, las ejecutamos. De lo contrario, salimos del bucle infinito y devolvemos la respuesta al usuario.

async def process_query(self, query):
  """Process a user query through Anthropic's Claude, handling tool calls iteratively."""
  messages = [{'role': 'user', 'content': query}]
  
  while True:
    response = self.anthropic.messages.create(
        max_tokens=2024,
        model='claude-3-7-sonnet-20250219', 
        tools=self.available_tools,
        messages=messages
    )
    
    assistant_content = []
    has_tool_use = False
    
    for content in response.content:
        if content.type == 'text':
            print(content.text)
            assistant_content.append(content)
        elif content.type == 'tool_use':
            has_tool_use = True
            assistant_content.append(content)
            messages.append({'role': 'assistant', 'content': assistant_content})
            
            # Execute the tool call
            session = self.sessions.get(content.name)
            if not session:
                print(f"Tool '{content.name}' not found.")
                break
                
            result = await session.call_tool(content.name, arguments=content.input)
            messages.append({
                "role": "user", 
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": content.id,
                    "content": result.content
                }]
            })
      
      if not has_tool_use:
          break

Entonces, ahora hemos cubierto completamente cómo funciona el chatbot MCP en el capó. Ahora, es hora de probarlo en acción. Puede ejecutarlo desde la interfaz de línea de comando con el siguiente comando.

python mcp_client_example_base.py

Cuando ejecute el chatbot, primero verá el siguiente mensaje de introducción que describe las opciones potenciales:

MCP Chatbot Started!
Commands:
  quit                           - Exit the chatbot
  @periods                       - Show available changelog periods
  @<period>                      - View changelog for specific period
  /tools                         - List available tools
  /tool <name> <arg1=value1>     - Execute a tool with arguments
  /prompts                       - List available prompts
  /prompt <name> <arg1=value1>   - Execute a prompt with arguments

A partir de ahí, puede probar diferentes comandos, por ejemplo,

• Llame a la herramienta para enumerar las bases de datos disponibles en el DB
• Lista todas las indicaciones disponibles
• Use la plantilla de solicitud, llamándola así /prompt sql_query_prompt question=”How many customers did we have in May 2024?”.

Finalmente, puedo terminar tu chat escribiendo quit.

Query: /tool list_databases
[07/02/25 18:27:28] INFO     Processing request of type CallToolRequest                server.py:619
Tool 'list_databases' result:
[TextContent(type='text', text='INFORMATION_SCHEMA\ndatasets\ndefault\necommerce\necommerce_db\ninformation_schema\nsystem\n', annotations=None, meta=None)]

Query: /prompts
Available prompts:
- sql_query_prompt: Create a SQL query prompt
  Arguments:
    - question

Query: /prompt sql_query_prompt question="How many customers did we have in May 2024?"
[07/02/25 18:28:21] INFO     Processing request of type GetPromptRequest               server.py:619
Executing prompt 'sql_query_prompt'...
I'll create a SQL query to find the number of customers in May 2024.
[07/02/25 18:28:25] INFO     Processing request of type CallToolRequest                server.py:619
Based on the query results, here's the final SQL query:
```sql
select uniqExact(user_id) as customer_count
from ecommerce.sessions
where toStartOfMonth(action_date) = '2024-05-01'
format TabSeparatedWithNames
```
Query: /tool execute_sql_query query="select uniqExact(user_id) as customer_count from ecommerce.sessions where toStartOfMonth(action_date) = '2024-05-01' format TabSeparatedWithNames"
I'll help you execute this SQL query to get the unique customer count for May 2024. Let me run this for you.
[07/02/25 18:30:09] INFO     Processing request of type CallToolRequest                server.py:619
The query has been executed successfully. The results show that there were 246,852 unique customers (unique user_ids) in May 2024 based on the ecommerce.sessions table.

Query: quit

¡Se ve muy bien! ¡Nuestra versión básica funciona bien! Ahora, es hora de llevarlo un paso más allá y hacer que nuestro chatbot sea más inteligente enseñándolo para sugerir indicaciones relevantes sobre la marcha basada en la aportación del cliente.

Sugerencias rápidas

En la práctica, sugerir plantillas de inmediato que mejor coincidan con la tarea del usuario puede ser increíblemente útil. En este momento, los usuarios de nuestro chatbot ya deben conocer las indicaciones disponibles o al menos ser lo suficientemente curiosas como para explorarlos por su cuenta para beneficiarse de lo que hemos creado. Al agregar una función de sugerencias rápidas, podemos hacer este descubrimiento para nuestros usuarios y hacer que nuestro chatbot sea significativamente más conveniente y fácil de usar.

Lloremos las formas de agregar esta funcionalidad. Abordaría esta característica de la siguiente manera:

Evalúe la relevancia de las indicaciones utilizando el LLM. Itere a través de todas las plantillas de inmediato disponibles y, para cada una, evalúe si el aviso es una buena coincidencia para la consulta del usuario.

Sugerir un indicador de coincidencia para el usuario. Si encontramos la plantilla de solicitud relevante, compártala con el usuario y pregunte si desea ejecutarla.

Fusionar la plantilla de solicitud con la entrada del usuario. Si el usuario acepta, combine el mensaje seleccionado con la consulta original. Dado que las plantillas de inmediato tienen marcadores de posición, es posible que necesitemos el LLM para completarlas. Una vez que hayamos fusionado la plantilla de inmediato con la consulta del usuario, tendremos un mensaje actualizado listo para enviar al LLM.

Agregaremos esta lógica al process_query función. Gracias a nuestro diseño modular, es bastante fácil agregar esta mejora sin interrumpir el resto del código.

Comencemos implementando una función para encontrar la plantilla de inmediato más relevante. Utilizaremos el LLM para evaluar cada indicador y asignarlo un puntaje de relevancia de 0 a 5. Después de eso, filtraremos cualquier indicación con una puntuación de 2 o menos y devolveremos solo la más relevante (la que tiene el puntaje de relevancia más alto entre los resultados restantes).

async def _find_matching_prompt(self, query):
  """Find a matching prompt for the given query using LLM evaluation."""
  if not self.available_prompts:
    return None
  
  # Use LLM to evaluate prompt relevance
  prompt_scores = []
  
  for prompt in self.available_prompts:
    # Create evaluation prompt for the LLM
    evaluation_prompt = f"""
You are an expert at evaluating whether a prompt template is relevant for a user query.

User Query: "{query}"

Prompt Template:
- Name: {prompt['name']}
- Description: {prompt['description']}

Rate the relevance of this prompt template for the user query on a scale of 0-5:
- 0: Completely irrelevant
- 1: Slightly relevant
- 2: Somewhat relevant  
- 3: Moderately relevant
- 4: Highly relevant
- 5: Perfect match

Consider:
- Does the prompt template address the user's intent?
- Would using this prompt template provide a better response than a generic query?
- Are the topics and context aligned?

Respond with only a single number (0-5) and no other text.
"""
      
    try:
      response = self.anthropic.messages.create(
          max_tokens=10,
          model='claude-3-7-sonnet-20250219',
          messages=[{'role': 'user', 'content': evaluation_prompt}]
      )
      
      # Extract the score from the response
      score_text = response.content[0].text.strip()
      score = int(score_text)
      
      if score >= 3:  # Only consider prompts with score >= 3
          prompt_scores.append((prompt, score))
            
    except Exception as e:
        print(f"Error evaluating prompt {prompt['name']}: {e}")
        continue
  
  # Return the prompt with the highest score
  if prompt_scores:
      best_prompt, best_score = max(prompt_scores, key=lambda x: x[1])
      return best_prompt
  
  return None

La siguiente función que necesitamos implementar es una que combine la plantilla de solicitud seleccionada con la entrada del usuario. Confiaremos en el LLM para combinarlos de manera inteligente, llenando a todos los marcadores de posición según sea necesario.

async def _combine_prompt_with_query(self, prompt_name, user_query):
  """Use LLM to combine prompt template with user query."""
  # First, get the prompt template content
  session = self.sessions.get(prompt_name)
  if not session:
      print(f"Prompt '{prompt_name}' not found.")
      return None
  
  try:
      # Find the prompt definition to get its arguments
      prompt_def = None
      for prompt in self.available_prompts:
          if prompt['name'] == prompt_name:
              prompt_def = prompt
              break
      
      # Prepare arguments for the prompt template
      args = {}
      if prompt_def and prompt_def.get('arguments'):
          for arg in prompt_def['arguments']:
              arg_name = arg.name if hasattr(arg, 'name') else arg.get('name', '')
              if arg_name:
                  # Use placeholder format for arguments
                  args[arg_name] = '<' + str(arg_name) + '>'
      
      # Get the prompt template with arguments
      result = await session.get_prompt(prompt_name, arguments=args)
      if not result or not result.messages:
          print(f"Could not retrieve prompt template for '{prompt_name}'")
          return None
      
      prompt_content = result.messages[0].content
      prompt_text = self._extract_prompt_text(prompt_content)
      
      # Create combination prompt for the LLM
      combination_prompt = f"""
You are an expert at combining prompt templates with user queries to create optimized prompts.

Original User Query: "{user_query}"

Prompt Template:
{prompt_text}

Your task:
1. Analyze the user's query and the prompt template
2. Combine them intelligently to create a single, coherent prompt
3. Ensure the user's specific question/request is addressed within the context of the template
4. Maintain the structure and intent of the template while incorporating the user's query

Respond with only the combined prompt text, no explanations or additional text.
"""
      
      response = self.anthropic.messages.create(
          max_tokens=2048,
          model='claude-3-7-sonnet-20250219',
          messages=[{'role': 'user', 'content': combination_prompt}]
      )
      
      return response.content[0].text.strip()
      
  except Exception as e:
      print(f"Error combining prompt with query: {e}")
      return None

Entonces, simplemente actualizaremos el process_query Lógica Para verificar si hay indicaciones coincidentes, solicite al usuario confirmación y decida qué mensaje enviar a la LLM.

async def process_query(self, query):
  """Process a user query through Anthropic's Claude, handling tool calls iteratively."""
  # Check if there's a matching prompt first
  matching_prompt = await self._find_matching_prompt(query)
  
  if matching_prompt:
    print(f"Found matching prompt: {matching_prompt['name']}")
    print(f"Description: {matching_prompt['description']}")
    
    # Ask user if they want to use the prompt template
    use_prompt = input("Would you like to use this prompt template? (y/n): ").strip().lower()
    
    if use_prompt == 'y' or use_prompt == 'yes':
        print("Combining prompt template with your query...")
        
        # Use LLM to combine prompt template with user query
        combined_prompt = await self._combine_prompt_with_query(matching_prompt['name'], query)
        
        if combined_prompt:
            print(f"Combined prompt created. Processing...")
            # Process the combined prompt instead of the original query
            messages = [{'role': 'user', 'content': combined_prompt}]
        else:
            print("Failed to combine prompt template. Using original query.")
            messages = [{'role': 'user', 'content': query}]
    else:
        # Use original query if user doesn't want to use the prompt
        messages = [{'role': 'user', 'content': query}]
  else:
    # Process the original query if no matching prompt found
    messages = [{'role': 'user', 'content': query}]

  # print(messages)
  
  # Process the final query (either original or combined)
  while True:
    response = self.anthropic.messages.create(
        max_tokens=2024,
        model='claude-3-7-sonnet-20250219', 
        tools=self.available_tools,
        messages=messages
    )
    
    assistant_content = []
    has_tool_use = False
    
    for content in response.content:
      if content.type == 'text':
          print(content.text)
          assistant_content.append(content)
      elif content.type == 'tool_use':
          has_tool_use = True
          assistant_content.append(content)
          messages.append({'role': 'assistant', 'content': assistant_content})
          
          # Log tool call information
          print(f"\n[TOOL CALL] Tool: {content.name}")
          print(f"[TOOL CALL] Arguments: {json.dumps(content.input, indent=2)}")
          
          # Execute the tool call
          session = self.sessions.get(content.name)
          if not session:
              print(f"Tool '{content.name}' not found.")
              break
              
          result = await session.call_tool(content.name, arguments=content.input)
          
          # Log tool result
          print(f"[TOOL RESULT] Tool: {content.name}")
          print(f"[TOOL RESULT] Content: {result.content}")
          
          messages.append({
              "role": "user", 
              "content": [{
                  "type": "tool_result",
                  "tool_use_id": content.id,
                  "content": result.content
              }]
          })
      
    if not has_tool_use:
        break

Ahora, probemos nuestra versión actualizada con una pregunta sobre nuestros datos. Emocionadamente, el chatbot pudo encontrar el aviso correcto y usarlo para encontrar la respuesta correcta.

Query: How many customers did we have in May 2024?
Found matching prompt: sql_query_prompt
Description: Create a SQL query prompt
Would you like to use this prompt template? (y/n): y
Combining prompt template with your query...
[07/05/25 14:38:58] INFO     Processing request of type GetPromptRequest               server.py:619
Combined prompt created. Processing...
I'll write a query to count unique customers who had sessions in May 2024. Since this is a business metric, I'll exclude fraudulent sessions.

[TOOL CALL] Tool: execute_sql_query
[TOOL CALL] Arguments: {
  "query": "/* Count distinct users with non-fraudulent sessions in May 2024\n   Using uniqExact for precise user count\n   Filtering for May 2024 using toStartOfMonth and adding date range */\nSELECT \n    uniqExactIf(s.user_id, s.is_fraud = 0) AS active_customers_count\nFROM ecommerce.sessions s\nWHERE toStartOfMonth(action_date) = toDate('2024-05-01')\nFORMAT TabSeparatedWithNames"
}
[07/05/25 14:39:17] INFO     Processing request of type CallToolRequest                server.py:619
[TOOL RESULT] Tool: execute_sql_query
[TOOL RESULT] Content: [TextContent(type='text', text='active_customers_count\n245287\n', annotations=None, meta=None)]
The query shows we had 245,287 unique customers with legitimate (non-fraudulent) sessions in May 2024. Here's a breakdown of why I wrote the query this way:

1. Used uniqExactIf() to get precise count of unique users while excluding fraudulent sessions in one step
2. Used toStartOfMonth() to ensure we capture all days in May 2024
3. Specified the date format properly with toDate('2024-05-01')
4. Used TabSeparatedWithNames format as required
5. Provided a meaningful column alias

Would you like to see any variations of this analysis, such as including fraudulent sessions or breaking down the numbers by country?

Siempre es una buena idea probar ejemplos negativos también. En este caso, el chatbot se comporta como se esperaba y no sugiere un mensaje relacionado con SQL cuando se le da una pregunta no relacionada.

Query: How are you?
I should note that I'm an AI assistant focused on helping you work with the available tools, which include executing SQL queries, getting database/table information, and accessing GitHub PR data. I don't have a tool specifically for responding to personal questions.

I can help you:
- Query a ClickHouse database
- List databases and describe tables
- Get information about GitHub Pull Requests

What would you like to know about these areas?

Ahora que nuestro chatbot está en funcionamiento, estamos listos para concluir las cosas.

Bonificación: Cliente MCP rápido y fácil con Smolagents

Hemos analizado un código de bajo nivel que permite construir clientes MCP altamente personalizados, pero muchos casos de uso requieren solo funcionalidad básica. Entonces, decidí compartir con ustedes una implementación rápida y directa de escenarios cuando necesita solo las herramientas. Usaremos uno de mis marcos de agentes favoritos: Smolagents de Huggingface (He discutido este marco en detalle en Mi artículo anterior).

# needed imports
from smolagents import CodeAgent, DuckDuckGoSearchTool, LiteLLMModel, VisitWebpageTool, ToolCallingAgent, ToolCollection
from mcp import StdioServerParameters
import json
import os

# setting OpenAI APIKey 
with open('../../config.json') as f:
    config = json.loads(f.read())

os.environ["OPENAI_API_KEY"] = config['OPENAI_API_KEY']

# defining the LLM 
model = LiteLLMModel(
    model_id="openai/gpt-4o-mini",  
    max_tokens=2048
)

# configuration for the MCP server
server_parameters = StdioServerParameters(
    command="uv",
    args=[
        "--directory",
        "/path/to/github/mcp-analyst-toolkit/src/mcp_server",
        "run",
        "server.py"
    ],
    env={"GITHUB_TOKEN": "github_<your_token>"},
)

# prompt 
CLICKHOUSE_PROMPT_TEMPLATE = """
You are a senior data analyst with more than 10 years of experience writing complex SQL queries, specifically optimized for ClickHouse to answer user questions.

## Database Schema

You are working with an e-commerce analytics database containing the following tables:

### Table: ecommerce.users 
**Description:** Customer information for the online shop
**Primary Key:** user_id
**Fields:** 
- user_id (Int64) - Unique customer identifier (e.g., 1000004, 3000004)
- country (String) - Customer's country of residence (e.g., "Netherlands", "United Kingdom")
- is_active (Int8) - Customer status: 1 = active, 0 = inactive
- age (Int32) - Customer age in full years (e.g., 31, 72)

### Table: ecommerce.sessions 
**Description:** User session data and transaction records
**Primary Key:** session_id
**Foreign Key:** user_id (references ecommerce.users.user_id)
**Fields:** 
- user_id (Int64) - Customer identifier linking to users table (e.g., 1000004, 3000004)
- session_id (Int64) - Unique session identifier (e.g., 106, 1023)
- action_date (Date) - Session start date (e.g., "2021-01-03", "2024-12-02")
- session_duration (Int32) - Session duration in seconds (e.g., 125, 49)
- os (String) - Operating system used (e.g., "Windows", "Android", "iOS", "MacOS")
- browser (String) - Browser used (e.g., "Chrome", "Safari", "Firefox", "Edge")
- is_fraud (Int8) - Fraud indicator: 1 = fraudulent session, 0 = legitimate
- revenue (Float64) - Purchase amount in USD (0.0 for non-purchase sessions, >0 for purchases)

## ClickHouse-Specific Guidelines

1. **Use ClickHouse-optimized functions:**
   - uniqExact() for precise unique counts
   - uniqExactIf() for conditional unique counts
   - quantile() functions for percentiles
   - Date functions: toStartOfMonth(), toStartOfYear(), today()

2. **Query formatting requirements:**
   - Always end queries with "format TabSeparatedWithNames"
   - Use meaningful column aliases
   - Use proper JOIN syntax when combining tables
   - Wrap date literals in quotes (e.g., '2024-01-01')

3. **Performance considerations:**
   - Use appropriate WHERE clauses to filter data
   - Consider using HAVING for post-aggregation filtering
   - Use LIMIT when finding top/bottom results

4. **Data interpretation:**
   - revenue > 0 indicates a purchase session
   - revenue = 0 indicates a browsing session without purchase
   - is_fraud = 1 sessions should typically be excluded from business metrics unless specifically analyzing fraud

## Response Format
Provide only the SQL query as your answer. Include brief reasoning in comments if the query logic is complex. 

## Examples

**Question:** How many customers made purchase in December 2024?
**Answer:** select uniqExact(user_id) as customers from ecommerce.sessions where toStartOfMonth(action_date) = '2024-12-01' and revenue > 0 format TabSeparatedWithNames

**Question:** What was the fraud rate in 2023, expressed as a percentage?
**Answer:** select 100 * uniqExactIf(user_id, is_fraud = 1) / uniqExact(user_id) as fraud_rate from ecommerce.sessions where toStartOfYear(action_date) = '2023-01-01' format TabSeparatedWithNames

**Question:** What was the share of users using Windows yesterday?
**Answer:** select 100 * uniqExactIf(user_id, os = 'Windows') / uniqExact(user_id) as windows_share from ecommerce.sessions where action_date = today() - 1 format TabSeparatedWithNames

**Question:** What was the revenue from Dutch users aged 55 and older in December 2024?
**Answer:** select sum(s.revenue) as total_revenue from ecommerce.sessions as s inner join ecommerce.users as u on s.user_id = u.user_id where u.country = 'Netherlands' and u.age >= 55 and toStartOfMonth(s.action_date) = '2024-12-01' format TabSeparatedWithNames

**Question:** What are the median and interquartile range (IQR) of purchase revenue for each country?
**Answer:** select country, median(revenue) as median_revenue, quantile(0.25)(revenue) as q25_revenue, quantile(0.75)(revenue) as q75_revenue from ecommerce.sessions as s inner join ecommerce.users as u on u.user_id = s.user_id where revenue > 0 group by country format TabSeparatedWithNames

**Question:** What is the average number of days between the first session and the first purchase for users who made at least one purchase?
**Answer:** select avg(first_purchase - first_action_date) as avg_days_to_purchase from (select user_id, min(action_date) as first_action_date, minIf(action_date, revenue > 0) as first_purchase, max(revenue) as max_revenue from ecommerce.sessions group by user_id) where max_revenue > 0 format TabSeparatedWithNames

**Question:** What is the number of sessions in December 2024, broken down by operating systems, including the totals?
**Answer:** select os, uniqExact(session_id) as session_count from ecommerce.sessions where toStartOfMonth(action_date) = '2024-12-01' group by os with totals format TabSeparatedWithNames

**Question:** Do we have customers who used multiple browsers during 2024? If so, please calculate the number of customers for each combination of browsers.
**Answer:** select browsers, count(*) as customer_count from (select user_id, arrayStringConcat(arraySort(groupArray(distinct browser)), ', ') as browsers from ecommerce.sessions where toStartOfYear(action_date) = '2024-01-01' group by user_id) group by browsers order by customer_count desc format TabSeparatedWithNames

**Question:** Which browser has the highest share of fraud users?
**Answer:** select browser, 100 * uniqExactIf(user_id, is_fraud = 1) / uniqExact(user_id) as fraud_rate from ecommerce.sessions group by browser order by fraud_rate desc limit 1 format TabSeparatedWithNames

**Question:** Which country had the highest number of first-time users in 2024?
**Answer:** select country, count(distinct user_id) as new_users from (select user_id, min(action_date) as first_date from ecommerce.sessions group by user_id having toStartOfYear(first_date) = '2024-01-01') as t inner join ecommerce.users as u on t.user_id = u.user_id group by country order by new_users desc limit 1 format TabSeparatedWithNames

---

**Your Task:** Using all the provided information above, write a ClickHouse SQL query to answer the following customer question: 
{question}
"""

with ToolCollection.from_mcp(server_parameters, trust_remote_code=True) as tool_collection:
  agent = ToolCallingAgent(tools=[*tool_collection.tools], model=model)
  prompt = CLICKHOUSE_PROMPT_TEMPLATE.format(
      question = 'How many customers did we have in May 2024?'
  )
  response = agent.run(prompt)

Como resultado, recibimos la respuesta correcta.

Si no necesita mucha personalización o integración con indicaciones y recursos, esta implementación es definitivamente el camino a seguir.

Resumen

En este artículo, creamos un chatbot que se integra con los servidores MCP y aprovecha todos los beneficios de la estandarización para acceder a herramientas, indicaciones y recursos sin problemas.

Comenzamos con una implementación básica capaz de enumerar y acceder a las capacidades de MCP. Luego, mejoramos nuestro chatbot con una característica inteligente que sugiere plantillas de inmediato relevantes para los usuarios en función de su entrada. Esto hace que nuestro producto sea más intuitivo y fácil de usar, especialmente para los usuarios que no están familiarizados con la biblioteca completa de indicaciones disponibles.

Para implementar nuestro chatbot, utilizamos un código de nivel relativamente bajo, lo que le brinda una mejor comprensión de cómo funciona el protocolo MCP debajo del capó y lo que sucede cuando usa herramientas de IA como Claude Desktop o cursor.

Como beneficio adicional, también discutimos la implementación de Smolagents que le permite implementar rápidamente un cliente MCP integrado con herramientas.

Gracias por leer. Espero que este artículo haya sido perspicaz. Recuerde el consejo de Einstein: “Lo importante es no dejar de cuestionar. La curiosidad tiene su propia razón para existir”. Que su curiosidad lo lleve a su próxima gran visión.

Referencia

Este artículo está inspirado en el “MCP: construya aplicaciones AI de contexto rico con antrópico“ curso corto de Deeplearning.ai.