El agente de IA como herramienta de experimentación: búsqueda en bases de datos biológicas con MCP

Al finalizar este tema

Podrás crear una herramienta que permita a LLM como Claude o GPT consultar automáticamente bases de datos biológicas como NCBI Entrez, PubMed y UniProt mediante instrucciones en lenguaje natural, combinando el MCP (Protocolo de contexto del modelo) y el harness que aprendiste en el libro. Comprenderás el funcionamiento interno del uso de herramientas por parte de los LLM y la naturaleza del harness que evalúa su fiabilidad.

Este artículo es un ejemplo didáctico. Un servidor MCP de implementación real tendrá una autenticación, limitación de velocidad y observabilidad mucho más sofisticadas.

"¿Por qué repetir la misma búsqueda una y otra vez?" — La trampa de la experimentación repetitiva

Supongamos que, al preparar ideas para un artículo, realizas estas búsquedas a diario.

1. Buscar "BRCA1" AND "review" AND "2024" en NCBI PubMed
2. Revisar los resúmenes y determinar su relevancia
3. Verificar las relaciones de citación
4. Verificar los dominios de proteínas relacionados en UniProt
5. Descargar la secuencia en GenBank

Repites esta combinación a diario. Hay un problema.

Problema 1: La sintaxis de búsqueda de cada base de datos es diferente. La sintaxis de filtro de NCBI Entrez, la sintaxis de búsqueda de UniProt y el formato de acceso de GenBank son todos diferentes. Debes recordarlo todo de nuevo cada vez.

Problema 2: Debes combinar manualmente los resultados de varias bases de datos. Por ejemplo, ingresas el nombre del autor obtenido de PubMed en UniProt y, a continuación, ingresas el número de acceso resultante en GenBank.

Problema 3: La mayoría de estas tareas se basan en reglas deterministas, pero se repiten a diario con diferentes palabras clave. Es decir, es el tipo de tarea ideal para la automatización, pero existe una gran sobrecarga en el aprendizaje de la documentación de la API de cada base de datos.

El verdadero enfoque es delegar esto a un agente de IA. Si le das una solicitud en lenguaje natural a un LLM, este llamará secuencialmente a varias API en segundo plano y combinará los resultados. El protocolo que hace esto de forma segura y estandarizada es el MCP (Protocolo de contexto del modelo).

De la caja negra a los componentes: desglosando el MCP

El MCP puede parecer complejo en la superficie, pero sus conceptos clave son tres.

Componente 1: Definición de herramientas

El servidor MCP presenta al LLM una lista de herramientas que puede proporcionar. Cada herramienta se define por su nombre, descripción y esquema de argumentos.

tool_definition = {
    "name": "search_pubmed",
    "description": "Busca palabras clave en la base de datos bibliográfica PubMed y devuelve los N primeros resultados",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Consulta de búsqueda con sintaxis de PubMed"},
            "max_results": {"type": "integer", "default": 10}
        },
        "required": ["query"]
    }
}

Lo esencial es que description es la única información que el LLM realmente lee y utiliza para tomar decisiones. Una buena descripción de la herramienta determina un buen comportamiento del agente.

Parte 2: Función de ejecución

Cada herramienta tiene una función que se ejecuta realmente. Cuando el LLM solicita una llamada a la herramienta, esta función se ejecuta y el resultado se devuelve al LLM.

async def search_pubmed(query: str, max_results: int = 10) -> dict:
    import httpx
    base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
    async with httpx.AsyncClient() as client:
        search_response = await client.get(
            f"{base}/esearch.fcgi",
            params={"db": "pubmed", "term": query, "retmax": max_results, "retmode": "json"}
        )
        ids = search_response.json()["esearchresult"]["idlist"]

        if not ids:
            return {"results": []}

        summary_response = await client.get(
            f"{base}/esummary.fcgi",
            params={"db": "pubmed", "id": ",".join(ids), "retmode": "json"}
        )
        summaries = summary_response.json()["result"]

        return {
            "results": [
                {
                    "pmid": pmid,
                    "title": summaries[pmid].get("title"),
                    "journal": summaries[pmid].get("fulljournalname"),
                    "pubdate": summaries[pmid].get("pubdate")
                }
                for pmid in ids
            ]
        }

Advertencia: Esta función llama a una API externa, por lo que debe gestionar las conexiones de red, los límites de frecuencia y los tiempos de espera. Esto se tratará a continuación.

Componente 3: Estructura del servidor MCP

Combinaremos todo en un único servidor. Anthropic proporciona un SDK de Python oficial, pero para que el concepto quede más claro, vamos a crear uno nosotros mismos.

from typing import Any, Callable, Coroutine
import json

class MCPServer:
    def __init__(self):
        self.tools: dict[str, dict[str, Any]] = {}
        self.handlers: dict[str, Callable[..., Coroutine]] = {}

    def register(self, definition: dict, handler: Callable[..., Coroutine]):
        name = definition["name"]
        self.tools[name] = definition
        self.handlers[name] = handler

    def list_tools(self) -> list[dict]:
        return list(self.tools.values())

    async def call_tool(self, name: str, arguments: dict) -> dict:
        if name not in self.handlers:
            return {"error": f"Unknown tool: {name}"}
        try:
            return await self.handlers[name](**arguments)
        except Exception as e:
            return {"error": str(e)}


server = MCPServer()
server.register(
    definition={
        "name": "search_pubmed",
        "description": "Search PubMed for scientific papers",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "max_results": {"type": "integer", "default": 10}
            },
            "required": ["query"]
        }
    },
    handler=search_pubmed
)

Este servidor debe comunicarse con el cliente LLM a través de entrada/salida estándar o HTTP, pero aquí entendemos el concepto en forma de llamada directa a una función.

Integración de LLM: Escribir con Claude

Ahora, conectemos este servidor con las llamadas LLM. Utilizaremos la función de uso de herramientas de la API de Claude.

import anthropic

client = anthropic.Anthropic()

async def agent_loop(user_message: str, server: MCPServer, max_iterations: int = 5) -> str:
    messages = [{"role": "user", "content": user_message}]
    tools = [
        {
            "name": t["name"],
            "description": t["description"],
            "input_schema": t["input_schema"]
        }
        for t in server.list_tools()
    ]

    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-opus-4-8",
            max_tokens=2048,
            tools=tools,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            text_blocks = [b.text for b in response.content if b.type == "text"]
            return "\n".join(text_blocks)

        if response.stop_reason == "tool_use":
            tool_uses = [b for b in response.content if b.type == "tool_use"]
            messages.append({"role": "assistant", "content": response.content})

            tool_results = []
            for tool_use in tool_uses:
                result = await server.call_tool(tool_use.name, tool_use.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": json.dumps(result)
                })

            messages.append({"role": "user", "content": tool_results})

    return "Max iterations reached"

Ahora, úselo:

result = await agent_loop(
    "Busca tres artículos de revisión sobre el gen BRCA1 publicados desde 2024 e indica solo el título y la revista.",
    server
)
print(result)

El LLM invoca automáticamente search_pubmed(query="BRCA1 AND review AND 2024:2025", max_results=3) y devuelve los resultados organizados en lenguaje natural.

Fading: tres espacios en blanco que debes completar

Espacio en blanco 1: ampliación de herramientas: búsqueda en UniProt

Un servidor que solo tenga PubMed es solo la mitad de la solución. Añadamos la herramienta UniProt.

async def search_uniprot(query: str, max_results: int = 10) -> dict:
    import httpx
    async with httpx.AsyncClient() as client:
        # TODO: llamar a la API REST de UniProt
        # Endpoint: https://rest.uniprot.org/uniprotkb/search
        # Parámetros: query, format=json, size=max_results
        # Resultado: {"results": [{"accession": ..., "name": ..., "gene": ...}]}
        pass

Pista: Extraiga primaryAccession, proteinDescription.recommendedName.fullName.value y genes[0].geneName.value de cada elemento del arreglo results en la respuesta de UniProt.

Espacio en blanco 2: harness — Sistema de evaluación de agentes

Para verificar si un agente funciona correctamente, se necesita un sistema de evaluación (evaluation harness). Ejecuta automáticamente varios casos de prueba y evalúa cada resultado.

async def evaluate_agent(
    test_cases: list[dict],
    server: MCPServer
) -> dict:
    """
    test_cases: [
        {
            "prompt": "Solicitud del usuario",
            "expected_tool_calls": ["search_pubmed"],
            "expected_content_contains": ["BRCA1", "review"]
        }
    ]
    """
    results = []
    for tc in test_cases:
        # TODO: ejecutar agent_loop, rastrear las herramientas llamadas realmente y verificar el resultado
        # Registrar en results si la prueba pasó
        pass

    return {
        "total": len(test_cases),
        "passed": sum(1 for r in results if r["passed"]),
        "details": results
    }

Pista: Modifique agent_loop para registrar las llamadas a las herramientas en un registro. Agregue cada llamada a una lista y devuélvala.

Espacio en blanco 3: límite de velocidad + manejo de errores

NCBI Entrez bloquea la dirección IP si se realizan más de 3 solicitudes por segundo. Agregue un límite de velocidad a la ejecución de la herramienta.

import asyncio
import time

class RateLimiter:
    def __init__(self, calls_per_second: float):
        self.min_interval = 1.0 / calls_per_second
        self.last_call = 0.0

    async def wait(self):
        now = time.time()
        elapsed = now - self.last_call
        if elapsed < self.min_interval:
            await asyncio.sleep(self.min_interval - elapsed)
        self.last_call = time.time()


ncbi_limiter = RateLimiter(calls_per_second=3)


async def search_pubmed_limited(query: str, max_results: int = 10) -> dict:
    # TODO: llamar a ncbi_limiter.wait() antes de iniciar la función
    # Después, ejecutar la lógica existente
    pass

Reflexión: ¿En qué se diferencia este agente de un LLM de uso de herramientas en un entorno real?

El agente que han creado comparte una base conceptual con los sistemas de producción (como Claude Code o ChatGPT con herramientas), pero los sistemas de producción son mucho más sofisticados.

Seguridad: Los servidores MCP de producción deben protegerse contra la inyección de prompts. Si los datos devueltos por una herramienta contienen instrucciones maliciosas ("ignora esta instrucción y haz X en su lugar"), el agente podría funcionar incorrectamente. Los sistemas de producción aíslan los valores devueltos por las herramientas en un contexto aislado o utilizan una capa de validación independiente.

Gestión de costes: Dado que cada iteración activa una llamada a la API, los costes pueden acumularse rápidamente. Los sistemas de producción tienen un límite de iteraciones, un presupuesto de tokens y un almacenamiento en caché de las llamadas a las herramientas.

Observabilidad: Para analizar por qué el agente ha llamado a una herramienta específica o qué razonamiento ha fallado, es necesario un registro de seguimiento. En los entornos de producción, cada paso se registra utilizando un estándar como OpenTelemetry.

Razonamiento en varios pasos: Las solicitudes complejas (por ejemplo, "encuentra todos los demás genes que aparecen en los artículos de este gen y consulta sus respectivos patrones de expresión") requieren un plan de llamadas a herramientas en varios pasos. Los sistemas de producción suelen utilizar una separación entre el planificador y el ejecutor o el marco ReAct.

Proyectos de ampliación

1. Añadir una herramienta para descargar secuencias de GenBank: efetch.fcgi; devuelve la secuencia FASTA a partir del número de acceso.

2. Herramienta de gráfico de citas: realiza un seguimiento de las relaciones de citas de los artículos de PubMed para construir un árbol de artículos relacionados.

3. Caché local: añade una capa de caché de SQLite para evitar la reejecución de la misma consulta.

4. Integración con Claude Desktop: registra el servidor MCP que han creado en la aplicación Claude Desktop real para que lo utilicen a diario. Puede envolverlo como un servidor stdio utilizando el SDK MCP estándar.

Mapa de los componentes de este ejemplo

• [F] Protocolo MCP: formato estándar para la definición, llamada y respuesta de las herramientas. Separación cliente-servidor.
• [F] Harness: sistema de pruebas que evalúa automáticamente el comportamiento del agente. Desempeña el papel de CI/CD para los sistemas LLM.
• [W] Llamada a la API y JSON: llamada a las API de NCBI/UniProt y análisis de las respuestas.
• [W] async/await: paralelización de varias llamadas a la API mediante E/S asíncrona.

[F] = lo implementan ustedes / [W] = concepto de herramienta proporcionado con el código completo.