Inicio / Artículos

mcplangchainragagents

IA agente con LangChain — Parte 6: RAG agente impulsado por MCP

Parte 6 de la IA agente con LangChain: RAG agente impulsado por MCP, donde las herramientas de recuperación provienen de un servidor MCP y LangGraph orquesta el flujo del agente.

Editor: REACTAPP.TOP12 min de lectura2686 palabras

Fuentes: Agentic AI with LangChain — Part 6: MCP Powered Agentic RAG with LangGraph

Esta guía reconstruye el proceso desde las materias primas hasta un sistema funcional para: IA agente con LangChain — Parte 6: RAG agente impulsado por MCP con LangGraph. El enfoque está en pasos operativos, verificaciones explícitas y código que se puede incorporar directamente a un repositorio sin tener que adivinar la intención. Para obtener una visión general, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.

Conceptos técnicos

Al trabajar con los Conceptos Técnicos, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo de tokens o consultas junto a los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la fase de demostración a entornos compartidos. Mida el rendimiento en un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.

1. Configuraciones del Sistema

Al trabajar en la sección 1. Configuraciones del sistema, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Mida la tasa de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. Estructura del proyecto

Al trabajar en la sección 2. Estructura del proyecto, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Documente junto con ello el camino óptimo y el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no de mejoras posteriores. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.

project/
├── rag_process/       # Pinecone RAG operations
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── graphs             # implement ReAct and Reflection StateGraph
│   ├── nodes.py       # graph nodes. Defined as Python functions
│   ├── graph.py       # graph workflow for MCP in SES mode
│   ├── graph_stdio.py # graph workflow for MCP in STDIO mode
│   ├── consts.py      # define node names as constants
│   └── state.py       # typed dictionary to store graph states
│

├── mcp_client_sse.py     # mcp client in sse mode
├── mcp_client_stdio.py   # mcp client in stdio mode
├── server_sse.py         # mcp server in sse mode
│── server_stdio.py       # mcp server in stdio mode
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── utils.py            # utility extracting question/docs from message list
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

3. MCP (Model Context Protocol)

Al trabajar con el 3. MCP (Model Context Protocol), anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestas las futuras modificaciones del código. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Almacene en caché las instrucciones del sistema estables y los esquemas de las herramientas. Reenviar un preámbulo idéntico es una causa común de desperdicio.

3.1. Concepto

Al trabajar en la sección 3.1. Concepto, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestos los cambios posteriores en el código. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina las comprobaciones de éxito y evite completaciones parciales silenciosas. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.

3.2 Arquitectura MCP

Al trabajar en la Arquitectura MCP 3.2, primero anote el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos. Mida el rendimiento en un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.

3.3. Modos de transporte MCP

Al trabajar en la sección 3.3. Modos de transporte MCP, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Mida la tasa de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.

3.3.1. Modo STDIO

Al trabajar en la sección 3.3.1. Modo STDIO, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Documente tanto la ruta de éxito como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras adicionales. Mida el rendimiento de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente. Al trabajar en la sección 3.3.1. Modo STDIO, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina las verificaciones de éxito y evite completaciones parciales silenciosas.

3.3.2. Modo SSE

3.3.2. El modo SSE funciona mejor cuando se trata como una superficie medible. Capture un registro exitoso, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la versión de demostración a entornos compartidos. Separe la política de fragmentación de la política de recuperación; cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.

4. Implementación de MCP en modo STDIO

  1. La implementación de MCP en modo STDIO funciona mejor cuando se trata como una superficie medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Separe la política de fragmentación de la política de recuperación. Cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.

4.1. Servidor MCP

4.1. El MCP Server funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documente tanto el camino óptimo como el de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Separe la política de fragmentación de la política de recuperación. Cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad. 4.1. El MCP Server funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Considere esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.

# initialize an STDIO FastMCP instance
mcp = FastMCP("Search Server")

# create mcp tools
@mcp.tool()
async def search_agentic_docs(query: str) -> str:
    """
    Good for questions about generative agents, prompt engineering, and adversarial attacks.
    Pass a natural language search query to retrieve context from the agentic ai database.
    """
    # Simply invoke your existing vector service retriever
    docs = await vector_service.retriever.ainvoke(query)

    # Flatten the document content into a single string for the LLM
    return "\n\n".join([d.page_content for d in docs])

# start MCP server in stdio mode
if __name__ == "__main__":
   mcp.run(transport="stdio")

4.2. MCP Client

Para el MCP Client 4.2, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo de tokens o consultas junto con los resultados funcionales. La visibilidad temprana del costo evita facturas inesperadas cuando la tarea pasa de entornos de demostración a entornos compartidos. Cite los pasajes que realmente sustentan la respuesta; sin citas, los operadores no pueden distinguir entre alucinaciones y lagunas en el indexado.

# define location of server file
BASE_DIR = Path(__file__).parent
SERVER_SCRIPT = BASE_DIR / "server.py"

# Local server running directly via Python over stdio
SERVER_CONFIG = {

    "local_search_server": {
        "transport": "stdio",
        "command": "python",
        "args": [str(SERVER_SCRIPT)],
    },
}


# create a MCP client and return
def get_mcp_client() :
    """Works for stdio, sse, or a mix of both!"""
    client = MultiServerMCPClient(SERVER_CONFIG)
    return client

4.3. Uso de herramientas MCP en el agente

Para 4.3. Al utilizar herramientas MCP en el agente, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el sistema. Mencione las secciones que realmente sirvieron como base para la respuesta. Sin citaciones, los operadores no podrán distinguir entre alucinaciones y lagunas en el indexado.

from mcp_client_stdio import get_mcp_client


async def build_graph():
    """
    build StateGraph
    :return:
    """
    # initialize a MCP client instance
    client = get_mcp_client()

    # retrieve MCP tools
    mcp_tools = await client.get_tools()

    # create an llM
    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
    tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools

    # bind tools to ToolNode and llm
    tool_node = ToolNode(tools)
    workflow = StateGraph(GraphState)
    llm_with_tools = llm.bind_tools(tools)

4.4. Pruebas del flujo de trabajo STDIO

Para 4.4. Pruebas del flujo de trabajo STDIO, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Cite los pasajes que realmente sustentan la respuesta. Sin citas, los operadores no pueden distinguir entre alucinaciones y lagunas en el indexado. Para 4.4. Pruebas del flujo de trabajo STDIO, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas.

question_1 = "what is prompt engineer ?"

async def main():
    print("hello ReAct agent by Langraph")
    message = HumanMessage(content=question_1)
    app = await build_graph()
    res = await app.ainvoke({"messages": [message]})


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

5. Implementación de MCP en modo SSE

Al trabajar en la sección 5. Implementación de MCP en modo SSE, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestos los cambios posteriores en el código. Registre los tiempos de ejecución y el costo del token o la consulta junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos. Mida el rendimiento en un conjunto fijo de preguntas antes de ajustar los prompts. Cambiar constantemente los prompts rara vez soluciona un sistema de recuperación deficiente.

5.1. Servidor MCP

Al trabajar en la sección 5.1. Servidor MCP, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Mida la tasa de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.

# for SSE
mcp = FastMCP(
    name="My MCP Server",
    host="0.0.0.0",
    port=8001
)

# implement mcp tools here

if __name__ == "__main__":
    mcp.run(transport="sse")

5.2. Cliente MCP

Al trabajar en la sección 5.2. MCP Client, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Documente tanto la ruta de éxito como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Mida el rendimiento de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente. Al trabajar en la sección 5.2. MCP Client, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas.

# specify config of the SSE MCP server that the client will connect to
SERVER_CONFIG = {
    # config for mcp by sse transportation
    "fastapi_mcp_server": {
        "transport": "sse",
        "url": "http://localhost:8001/sse",
    }
}


# create a client
def get_mcp_client() :
    client = MultiServerMCPClient(SERVER_CONFIG)
    return client

5.3. Uso de herramientas MCP en StateGraph

5.3. El uso de las herramientas MCP en StateGraph funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando la ruta pasa de entornos de demostración a entornos compartidos. Separe la política de fragmentación de la política de recuperación; cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.

async def build_graph(mcp_tools: list):
    """
    build StateGraph
    :return:
    """

    # create an llm
    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
    tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools

    # bind tools to ToolNode and llm
    tool_node = ToolNode(tools)
    llm_with_tools = llm.bind_tools(tools)

5.4. Por qué la implementación SSE difiere de STDIO

5.4. Por qué la implementación de SSE difiere de STDIO: funciona mejor cuando se trata como una superficie medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema.

6. Integración del cliente SSE MCP con FastAPI

from mcp_client_sse import get_mcp_client

# CREATES a manager for FastAPI startup/shutdown
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):

    # create an MCP client, and retrieve MCP tools
    mcp_client = get_mcp_client()
    mcp_tools = await mcp_client.get_tools()

    # Compile graph dynamically with tools and store it on app state
    fastapi_app.state.agent = await build_graph(mcp_tools)

    # PAUSES here while FastAPI runs and handles requests
    yield


app = FastAPI(lifespan=lifespan)

7. Volver a construir StateGraph en la parte 5

7.1. Punto de extremo de FastAPI

@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(request: Request, body: RAGRequest=Body(...)) -> RAGResponse:
    try:
        # wrap input query in a HumanMessage
        message = HumanMessage(content=body.question)

        # loads the StateGraph agent from the request
        # this agent is created and maintained by lifespan context
        agent = request.app.state.agent

        # Async invokes the agent and returns the response
        generation = await agent.ainvoke({"messages": [message]})
        response = {"answer": generation.get("generation", "")}

        return RAGResponse(**response)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

7.2. Interfaz frontal de Streamlit

8. Herramientas gestionadas por MCP frente a herramientas locales de LangChain

9. Resumen final

Lista de verificación operativa

← Todos los artículosEnviar el brief