El protocolo MCP (Model Context Protocol) introduce el concepto de Sampling (muestreo) como el meacnismo fundamental que permite a las herramientas interactuar con los Modelos de Lenguaje Grande (LLM) para generar texto. En lugar de que el servidor MCP se conecte directamente a un LLM, la arquitectura delega esta tarea al cliente. Este diseño promueve la separación de responsabilidades: el servidor se enfoca exclusivamente en orquestar recursos y exponer herramientas, mientras que el cliente actúa como el puente que gestiona las credenciales, selecciona el modelo adecuado y ejecuta las peticiones de inferencia. A continuación, se detalla cómo configurar esta interacción utilizando la biblioteca FastMCP en Python.

Configuración del Servidor MCP

En el lado del servidor, se implementa una herramienta de clasificación de intenciones. Esta herramienta no posee acceso directo a un modelo de lenguaje; en su lugar, utiliza el contexto de ejecución para solicitar al cliente que realice el muestreo y devuelva la inferencia.

from fastmcp import FastMCP, Context
from mcp.types import SamplingMessage, TextContent
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

server = FastMCP("IntentClassifierServer")

@server.tool()
async def classify_intent(user_input: str, context: Context) -> dict:
    """
    Clasifica la intención principal del texto proporcionado.
    """
    system_instructions = (
        "Eres un clasificador de intenciones. Analiza el texto y devuelve "
        "únicamente una de las siguientes palabras: 'question', 'command', 'statement'."
    )

    logger.info(f"Procesando entrada para clasificación: {user_input}")

    # Solicitar al cliente que realice el sampling
    llm_response = await context.sample(
        messages=[
            SamplingMessage(
                role="user", 
                content=TextContent(type="text", text=user_input)
            )
        ],
        system_prompt=system_instructions
    )

    raw_intent = llm_response.text.strip().lower()

    # Validar y normalizar la respuesta
    valid_intents = {"question", "command", "statement"}
    final_intent = raw_intent if raw_intent in valid_intents else "statement"

    logger.info(f"Intención clasificada: {final_intent}")
    return {"input_text": user_input, "detected_intent": final_intent}

if __name__ == "__main__":
    server.run(transport="streamable-http", host="127.0.0.1", port=8080, show_banner=False)

El núcleo de esta implementación reside en la llamada a context.sample(). Cuando la herramienta requiere capacidades cognitivas, emite una solicitud a través del contexto. El servidor no necesita conocer la URL de la API del LLM ni las claves de acceso; simplemente delega la tarea de generación de texto al entorno del cliente.

Implementación del Cliente MCP

El cliente asume un rol dual: actúa como el anfitrión que consume las herramientas del servidor y como el proveedor de LLM que responde a las solicitudes de sampling. La clase MCPClientManager encapsula esta lógica, manejando tanto el bucle de conversación como el enrutamiento de las peticiones de inferencia.

import asyncio
import json
import logging
from typing import cast, Any

from fastmcp import Client
from fastmcp.client.sampling import SamplingMessage, SamplingParams
from mcp.shared.context import RequestContext
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageFunctionToolCall

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuración de entorno
LLM_API_KEY = "tu_api_key_aqui"
LLM_BASE_URL = "https://api.openai.com/v1"
LLM_MODEL = "gpt-3.5-turbo"

class MCPClientManager:
    def __init__(self, server_endpoint: str):
        self.endpoint = server_endpoint
        self.client = Client(server_endpoint, sampling_handler=self._process_sampling)
        self.llm_client = AsyncOpenAI(api_key=LLM_API_KEY, base_url=LLM_BASE_URL)
        self.conversation_history: list[dict[str, Any]] = []

    async def _process_sampling(
        self, 
        messages: list[SamplingMessage], 
        params: SamplingParams, 
        ctx: RequestContext
    ) -> str:
        """Manejador interno para las solicitudes de sampling del servidor."""
        prompt_messages = []
        if params.systemPrompt:
            prompt_messages.append({"role": "system", "content": params.systemPrompt})

        for msg in messages:
            text_content = msg.content.text if hasattr(msg.content, 'text') else str(msg.content)
            prompt_messages.append({"role": msg.role, "content": text_content})

        completion = await self.llm_client.chat.completions.create(
            model=LLM_MODEL,
            messages=prompt_messages,
            temperature=0.1
        )
        return completion.choices[0].message.content or ""

    async def _format_tools_for_llm(self, tools: list) -> list[dict]:
        """Convierte las herramientas MCP al formato de funciones de OpenAI."""
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.inputSchema,
                }
            }
            for t in tools
        ]

    async def execute_turn(self, user_query: str) -> str:
        """Ejecuta un turno de conversación, manejando llamadas a herramientas."""
        self.conversation_history.append({"role": "user", "content": user_query})
        accumulated_response = []

        async with self.client:
            mcp_tools = await self.client.list_tools()
            openai_tools = await self._format_tools_for_llm(mcp_tools)

            response = await self.llm_client.chat.completions.create(
                model=LLM_MODEL,
                messages=self.conversation_history,
                tools=openai_tools,
                temperature=0.2
            )

            assistant_message = response.choices[0].message
            if assistant_message.content:
                accumulated_response.append(assistant_message.content)

            while assistant_message.tool_calls:
                for tool_call in assistant_message.tool_calls:
                    if not hasattr(tool_call, "function"):
                        continue

                    func_call = cast(ChatCompletionMessageFunctionToolCall, tool_call)
                    tool_name = func_call.function.name
                    tool_args = json.loads(func_call.function.arguments)

                    if not self.client.is_connected():
                        raise ConnectionError("Conexión MCP perdida.")

                    tool_result = await self.client.call_tool(tool_name, tool_args)

                    self.conversation_history.append({
                        "role": "assistant",
                        "tool_calls": [{
                            "id": tool_call.id,
                            "type": "function",
                            "function": {"name": tool_name, "arguments": func_call.function.arguments}
                        }]
                    })
                    self.conversation_history.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(tool_result.content) if tool_result.content else ""
                    })

                follow_up = await self.llm_client.chat.completions.create(
                    model=LLM_MODEL,
                    messages=self.conversation_history,
                    tools=openai_tools,
                    temperature=0.2
                )
                assistant_message = follow_up.choices[0].message
                if assistant_message.content:
                    accumulated_response.append(assistant_message.content)

        return "\n".join(accumulated_response)

    async def run_cli(self):
        """Bucle principal de la interfaz de línea de comandos."""
        print("Sistema MCP iniciado. Escribe 'salir' para terminar.")
        while True:
            user_input = input("Usuario: ").strip()
            if user_input.lower() == 'salir':
                print("Finalizando sesión.")
                break
            if not user_input:
                continue

            try:
                result = await self.execute_turn(user_input)
                print(f"Asistente: {result}\n")
            except Exception as e:
                logger.error(f"Error durante la ejecución: {e}")

    async def close(self):
        await self.client.close()

async def main():
    manager = MCPClientManager(server_endpoint="http://127.0.0.1:8080/mcp")
    try:
        await manager.run_cli()
    finally:
        await manager.close()

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

La función clave aquí es _process_sampling. Cuando el servidor invoca context.sample(), el cliente intercepta la llamada a través de este manejador. El cliente construye el prompt, lo envía a la API de OpenAI y devuelve la respuesta generada al servidor, permitiendo que la herramienta continúe su ejecución con los datos inferidos.

Flujo de Ejecución e Interacción

Al ejecutar el sistema, la interacción se visualiza de la siguiente manera en la consola:

Sistema MCP iniciado. Escribe 'salir' para terminar.
Usuario: ¿Podrías decirme qué hora es?
Asistente: La intención detectada para tu consulta es 'question'.

Usuario: Reinicia el servidor de base de datos inmediatamente.
Asistente: He clasificado tu instrucción como 'command'.

Usuario: salir
Finalizando sesión.

Este resultado ilustra la secuencia exacta de eventos que ocurren en segundo plano:

1. El usuario ingresa un texto en la interfaz del cliente.
2. El cliente envía el texto al LLM principal, el cual decide invocar la herramienta classify_intent en el servidor.
3. El servidor recibe la petición y, en lugar de procesar el texto localmente, utiliza el mecanismo de Sampling para solicitar una segunda inferencia al cliente.
4. El cliente recibe la solicitud de sampling, consulta al LLM para obtener la etiqueta de intención y devuelve el resultado al servidor, el cual finalmente responde al usuario.