TL;DR
Los protocolos de contexto de modelo (MCP) son adaptadores estandarizados que permiten a las aplicaciones de IA conectarse a fuentes de datos y herramientas externas. Construimos un asistente de investigación utilizando MCP, LangChain, FastAPI y Streamlit. El sistema aprovecha dos servidores MCP para buscar artículos científicos en ArXiv y analizar, mediante docling OCR, el contenido para responder preguntas.
Los grandes modelos lingüísticos (LLM) están por todas partes, impulsando aplicaciones desde chatbots hasta complejos asistentes de IA. Sin embargo, entre bastidores, la integración de datos y funcionalidades externas suele implicar la creación repetida de herramientas, mensajes, datos, etc. de tipo repetitivo. Aquí es donde el Protocolo de Contexto de Modelo (MCP) tiene un impacto significativo, incluso impulsando software asistido por IA como Cursor o Claude Desktop, aprovechando el uso programático y personalizable de herramientas mientras se beneficia de las capacidades de razonamiento de LLM.
MCP está diseñado para tender un puente entre los LLM y los datos del mundo real. Actúa como un adaptador universal (a todo el mundo parece gustarle la analogía de MCP como un «puerto USB-C») para aplicaciones de IA, proporcionando una forma estandarizada y segura de exponer datos y funcionalidades externas a sus modelos. MCP separa las preocupaciones de proporcionar contexto de las interacciones LLM. Esto significa que puedes construir servidores MCP que expongan:
- Recursos: Endpoints de datos (similares a las peticiones GET) que cargan contexto en tu LLM.
- Herramientas: Endpoints funcionales (como peticiones POST) que permiten a tu LLM ejecutar código o desencadenar efectos secundarios.
- Peticiones: Plantillas reutilizables que estandarizan las interacciones.
En otras palabras, MCP elimina el trabajo extra de crear herramientas, etc para cada proyecto, permitiendo la integración out-of-the-box con el mundo exterior (o APIs de pago).
En este proyecto, utilizaremos MCP junto con LangChain, FastAPI y Streamlit para crear un sencillo pero potente asistente de investigación. La arquitectura se compone de tres componentes principales:
Servidores MCP:
- Servidor ArXiv: Proporciona las herramientas necesarias para la búsqueda y recuperación de artículos científicos.
- Servidor DocLing: Ofrece herramientas de lingüística documental para analizar textos y extraer ideas clave.
Servidor cliente FastAPI:
Actuando como capa de coordinación, este servidor agrega las funcionalidades de los servidores MCP. Implementa el agente asistente de investigación que aprovecha múltiples herramientas MCP y expone una API unificada para las interacciones con el cliente.
Interfaz de usuario Streamlit:
Esta interfaz web de fácil uso permite a los usuarios enviar consultas de investigación, ver los resultados de las búsquedas y analizar documentos sin problemas.
Y el resultado es algo parecido a esto:
docker-compose up --build
Y aparecerá una interfaz de usuario Streamlit como ésta:
Aquí nos centraremos en los componentes centrales que hacen funcionar nuestro asistente de investigación: los servidores MCP (que impulsan nuestra búsqueda en ArXiv y el análisis de documentos) y el servidor cliente FastAPI que coordina estos servicios. Nuestro diseño divide el backend en dos servidores MCP dedicados:
Servidor MCP de ArXiv
El servidor ArXiv se encarga de consultar y recuperar artículos científicos de ArXiv. Esto es lo que hace
- Modelo de artículo: Utiliza un modelo Pydantic para estructurar los resultados de la búsqueda.
- Función de búsqueda: Aprovecha la biblioteca arxiv para buscar artículos de investigación basados en una consulta.
- Exposición de la herramienta: Expone la función get_articles como una herramienta MCP (a través del decorador @mcp.tool) para que pueda ser invocada por el servidor cliente.
A continuación se muestra una versión simplificada del código:
# arxiv_server.py
import arxiv
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
import asyncio
from typing import Optional
# Initialize MCP server on port 8000
mcp = FastMCP("Research Article Provider", port=8000)
class Article(BaseModel):
title: str
summary: str
published_date: str
pdf_link: Optional[str]
@classmethod
def from_arxiv_result(cls, result: arxiv.Result) -> 'Article':
pdf_links = [str(i) for i in result.links if '/pdf/' in str(i)]
pdf_link = pdf_links[0] if pdf_links else None
return cls(
title=result.title,
summary=result.summary,
published_date=result.published.strftime('%Y-%m-%d'),
pdf_link=pdf_link
)
def __str__(self):
return (f'Title: {self.title}\nDate: {self.published_date}\n'
f'PDF Url: {self.pdf_link}\n\n' +
'\n'.join(self.summary.splitlines()[:3]) + '\n[...]')
def get_articles_content(query: str, max_results: int) -> list[Article]:
client = arxiv.Client()
search = arxiv.Search(query=query, max_results=max_results, sort_by=arxiv.SortCriterion.Relevance)
articles = map(lambda x: Article.from_arxiv_result(x), client.results(search))
articles_with_link = filter(lambda x: x.pdf_link is not None, articles)
return list(articles_with_link)
@mcp.tool(
name="search_arxiv",
description="Get `max_results` articles for a given search query on Arxiv."
)
async def get_articles(query: str, max_results: int) -> str:
print(f"Searching for '{query}'...")
articles = get_articles_content(query, max_results)
print(f"Found {len(articles)} articles.")
return '\n\n-------\n\n'.join(map(str, articles)).strip()
if __name__ == "__main__":
asyncio.run(mcp.run_sse_async())
¡Mira qué sencillo es exponer una herramienta usando MCP! Simplemente creamos una entidad FastMCP que representa al servidor (como FastAPI), y luego el decorador @mcp.tool registra automáticamente la función con el servidor MCP, haciéndola accesible al cliente-servidor. La función get_articles es la lógica real, buscando artículos en ArXiv basados en la relevancia, y devuelve una cadena formateada con los resultados. Esto es lo que parece:
Servidor MCP de Docling
El servidor Docling se encarga de la extracción y generación de vistas previas del contenido de artículos a partir de PDF. Los puntos clave incluyen:
- Conversión de documentos: Utiliza un DocumentConverter (de la biblioteca docling) para convertir el contenido PDF en Markdown.
- Previsualización de texto: Implementa un divisor de texto (usando RecursiveCharacterTextSplitter de LangChain y tiktoken) para tomar el primer trozo de texto para previsualizaciones rápidas.
- Herramientas expuestas: Proporciona dos herramientas: una para extraer el texto completo y otra para obtener una vista previa.
He aquí una versión del código del servidor DocLing:
# docling_server.py
from mcp.server.fastmcp import FastMCP
from docling.document_converter import DocumentConverter
from langchain_text_splitters import RecursiveCharacterTextSplitter
import tiktoken
import asyncio
# Initialize MCP server on port 8001
mcp = FastMCP("Research Article Extraction Provider", port=8001)
def get_article_content_str(article_url: str):
converter = DocumentConverter()
result = converter.convert(article_url)
return result.document.export_to_markdown()
def first_lines(text: str, chunk_size: int = 1536) -> str:
encoder = tiktoken.encoding_for_model('gpt-4')
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=0,
length_function=lambda x: len(encoder.encode(x)),
is_separator_regex=False,
)
return text_splitter.split_text(text)[0]
@mcp.tool(
name="extract_article_content",
description="Extracts the full text content from a research article PDF using OCR."
)
async def get_article_content(article_url: str) -> str:
return get_article_content_str(article_url).strip()
@mcp.tool(
name="get_article_preview",
description="Retrieves the first portion of a research article for quick previews."
)
async def get_article_first_lines(article_url: str, chunk_size: int = 1536) -> str:
articlecontent = get_article_content_str(article_url)
return first_lines(articlecontent.strip(), chunk_size).strip()
if __name__ == "__main__":
asyncio.run(mcp.run_sse_async())
Al igual que en el servidor ArXiv, definimos las herramientas y su lógica, y las exponemos utilizando el decorador @mcp.tool. Así es como se ve el uso de esta herramienta:
Servidor cliente
El servidor cliente, construido con FastAPI, actúa como capa de coordinación. Se conecta a ambos servidores MCP e integra sus funcionalidades utilizando un agente LangChain ReAct. Así es como funciona:
- Configuración del Cliente MCP: Se crea un MultiServerMCPClient con endpoints para los servidores ArXiv y DocLing.
- Agente ReAct: El agente (creado con create_react_agent) procesa la solicitud de investigación invocando de forma inteligente las herramientas disponibles.
- Punto final FastAPI: El endpoint /research recibe una consulta del usuario, la procesa a través del agente y devuelve la respuesta formateada.
A continuación se muestra un extracto del código cliente-servidor:
# client_server.py
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import AzureChatOpenAI
from dotenv import load_dotenv
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Dict, Any
import uvicorn
class ResearchRequest(BaseModel):
prompt: str
load_dotenv()
model = AzureChatOpenAI(azure_deployment=os.environ['AZURE_GPT_DEPLOYMENT'])
app = FastAPI(title="Research Assistant API")
async def process_prompt(prompt: str) -> Dict[str, Any]:
mcp_client = MultiServerMCPClient({
"arxiv": {
"url": "http://arxiv-server:8000/sse", # Docker service name
"transport": "sse",
},
"docling": {
"url": "http://docling-server:8001/sse", # Docker service name
"transport": "sse",
}
})
async with mcp_client as client:
agent = create_react_agent(model, client.get_tools())
response = await agent.ainvoke({"messages": prompt}, debug=False)
messages = [{i.type: i.content} for i in response['messages'] if i.content != '']
return {"messages": messages}
@app.post("/research")
async def research(request: ResearchRequest):
return await process_prompt(request.prompt)
if __name__ == '__main__':
uvicorn.run(app, host="0.0.0.0", port=8080)
Aquí envolvemos nuestro cliente-servidor que utiliza la librería langchain-mcp-adapters para transformar las herramientas MCP en herramientas LangChain. Sólo informamos al cliente cómo conectarse a los servidores e instanciamos el resto como un agente Langchain normal.
Interfaz de usuario de Streamlit
La aplicación Streamlit proporciona una interfaz sencilla y fácil de usar para interactuar con nuestro asistente de investigación. Dado que su código implica elementos e interacciones de interfaz de usuario web estándar, puede encontrar la implementación completa en el repositorio. En esencia, simplemente envía consultas de investigación al cliente-servidor y muestra los resultados.
Al modular el sistema con servidores MCP y coordinarlos a través de un cliente FastAPI utilizando el agente ReAct de LangChain, nos aseguramos de que el asistente de investigación siga siendo flexible y escalable. Este enfoque permite integrar fácilmente herramientas o fuentes de datos adicionales en el futuro.
Finalmente, podemos envolver todo usando Docker Compose y ¡tener un agente interactivo con una interfaz de usuario!
Siéntete libre de explorar el código completo en mi repositorio y experimentar con la configuración usando Docker Compose como se describe en el README del proyecto. ¡Feliz programación!