Reservar citas y reembolsar facturas desde un agente LangGraph con MCP
++
- 1State that survives a restart
- 2Hand work between agents
- 3Governed tools an agent can call
- 🏆Engineer the context, not the prompt
La Parte 1 le dio a un agente un estado que sobrevive a un reinicio, y una pausa donde un humano aprueba antes de que continúe. La Parte 2 puso un supervisor delante de un agente de agenda y uno de facturación, y repartió el trabajo entre ellos.
Ninguno de esos agentes agendó ni facturó nada. Lo comentaron. Cada herramienta que tenían era una función que tú escribiste en el mismo proceso, y las interesantes — buscar un cliente, ocupar un hueco, mover dinero — no existían.
Esta parte construye la caja de herramientas que esos dos roles necesitan, por MCP: un calendario, una lista de clientes, facturas y un reembolso. Para reducir piezas móviles apuntamos un agente a las cinco herramientas en lugar de reconstruir el supervisor de la Parte 2 — el agente las descubre en tiempo de ejecución en vez de estar cableado a ellas, las encadena para responder una pregunta, y escribe filas que puedes ir a comprobar en la base de datos después. Luego el reembolso lo detiene en seco y hace que un humano teclee el importe.
Repartir estas herramientas de vuelta entre el agente de agenda y el de facturación — de modo
que el de agenda no pueda ver issue_refund en absoluto — es el siguiente paso natural, y la
última sección explica cómo.
En julio la especificación de MCP eliminó las sesiones por completo. El día que ejecutamos esto, LangChain publicó el soporte para esa revisión en el paquete principal. Así que esta es una primera ejecución contra un protocolo de cinco semanas y un cliente del mismo día — que es justo por lo que tres cosas de aquí no están todavía en la documentación de nadie.
Una WEC Instance: 8 vCPU AMD EPYC 7601, 15 GB RAM, Ubuntu 22.04, Python 3.10. langchain
1.4.0, langchain-openai 1.6.0, langgraph 1.2.11, mcp 2.1.1, fastmcp 4.0.2 — todo
instalado el día de escribir esto. Modelos servidos a través del gateway LiteLLM de la
Parte 1 de la serie del gateway, que reenvía a
WEC Inference. langchain.mcp está en beta y avisa en cada importación; la API puede moverse.
Qué es MCP en realidad
Un servidor publica herramientas. Un cliente le pregunta qué herramientas existen. El modelo elige una y el cliente la llama. Ese es todo el protocolo — el valor está en que "preguntar qué herramientas existen" esté estandarizado, así que cualquier cliente puede usar cualquier servidor sin pegamento escrito para ese par concreto.
Tres palabras que necesitas:
Herramienta. Una función que el servidor expone, con una descripción y un esquema JSON para sus argumentos. La descripción es lo que el modelo lee para decidir si la llama; el esquema es lo que debe rellenar. Nada más de tu código llega al modelo.
Descubrimiento. El cliente pide tools/list y recibe ese catálogo. Tu agente aprende lo que
puede hacer en tiempo de ejecución en lugar de cuando lo escribiste.
Elicitación. Una herramienta que no puede terminar sin preguntarle algo a un humano — confirmar un borrado, aportar un parámetro que el modelo no tenía. Con la nueva especificación esto es una petición corriente que el cliente reintenta con la respuesta adjunta.
El viaje de ida y vuelta que sustituyó a la sesión
Con la especificación antigua, una herramienta que necesitaba datos mantenía la conexión
abierta y preguntaba por un canal de retorno. Con 2026-07-28 retorna, y el cliente la llama
otra vez con la respuesta adjunta. Dos peticiones completas en lugar de una de larga
duración:
La herramienta se ejecuta dos veces. Por eso su cuerpo empieza con
if ctx.input_responses is None — tiene que estar escrita para ser reentrada, no reanudada.
Nada queda abierto entre medias, así que la respuesta puede llegar después de un redespliegue, o
a otra réplica detrás de un balanceador.
Requisitos previos
- El entorno virtual y el acceso a WEC Inference de la Parte 1
- Una gateway LiteLLM con una clave virtual, o cualquier endpoint compatible con OpenAI al que puedas apuntar un modelo
- Python 3.10+
- Dos sesiones de terminal en la caja — el servidor bloquea una de ellas
Paso 1 — Instalar
El soporte de MCP ahora vive en el paquete langchain principal. Si tienes
langchain-mcp-adapters de antes, es justo lo que esto sustituye.
mkdir -p ~/mcp-tools && cd ~/mcp-tools && python3 -m venv .venv && ./.venv/bin/pip install -q "langchain[mcp]" langchain-openai httpx && ./.venv/bin/pip list | grep -iE 'langchain|langgraph|^mcp |fastmcp'
langchain[mcp] necesita 1.4.0 o superior — por debajo, MCPAdapter no existe.
mcp[cli] ya no existeLas guías antiguas te dicen que instales "mcp[cli]". En mcp 2.x ese extra ya no existe, y
pip falla con un muro de does not provide the extra 'cli' a lo largo de sesenta versiones
antes de rendirse con ResolutionImpossible. Parece un conflicto de dependencias; es un extra
eliminado.
Paso 2 — Algo sobre lo que las herramientas actúen
Los agentes necesitan datos que existan. SQLite, sembrado una vez — tres clientes, seis huecos (uno ya ocupado), cuatro facturas.
import sqlite3
db = sqlite3.connect("office.db")
db.executescript("""
DROP TABLE IF EXISTS customers; DROP TABLE IF EXISTS slots; DROP TABLE IF EXISTS invoices;
CREATE TABLE customers(id INTEGER PRIMARY KEY, name TEXT, email TEXT);
CREATE TABLE slots(id INTEGER PRIMARY KEY, day TEXT, time TEXT, customer_id INTEGER);
CREATE TABLE invoices(id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL, status TEXT);
INSERT INTO customers(name,email) VALUES
('Maria Alvarez','maria@example.com'),
('Tomas Reis','tomas@example.com'),
('Priya Nair','priya@example.com');
INSERT INTO slots(day,time,customer_id) VALUES
('2026-09-04','09:00',NULL),('2026-09-04','11:00',NULL),('2026-09-04','14:00',2),
('2026-09-05','09:00',NULL),('2026-09-05','11:00',NULL),('2026-09-05','16:00',NULL);
INSERT INTO invoices(customer_id,amount,status) VALUES
(1,240.00,'paid'),(1,80.00,'open'),(2,150.00,'paid'),(3,410.00,'open');
""")
db.commit()
Dos detalles cargan peso. Maria tiene una factura pagada y otra abierta, así que "reembolsa la factura de Maria" es ambiguo y el agente tiene que resolverlo. Tomas ya ocupa el viernes a las 14:00, así que la disponibilidad es una pregunta real y no "todo".
Paso 3 — La caja de herramientas
Cinco herramientas: tres de lectura, una de escritura, y una que no puede terminar sin un humano.
import os, sqlite3
from fastmcp import FastMCP, Context
from mcp.types import InputRequiredResult, ElicitRequest, ElicitRequestFormParams
DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "office.db")
mcp = FastMCP("office-tools")
def q(sql, args=()):
con = sqlite3.connect(DB); con.row_factory = sqlite3.Row
try:
rows = con.execute(sql, args).fetchall(); con.commit()
return [dict(r) for r in rows]
finally:
con.close()
def w(sql, args=()):
con = sqlite3.connect(DB)
try:
cur = con.execute(sql, args); con.commit(); return cur.rowcount
finally:
con.close()
@mcp.tool
async def find_customer(query: str) -> list[dict]:
"""Find customers whose name or email matches the query."""
like = f"%{query}%"
return q("SELECT id, name, email FROM customers WHERE name LIKE ? OR email LIKE ?", (like, like))
@mcp.tool
async def list_availability(day: str) -> list[dict]:
"""List free appointment slots on a given day, formatted YYYY-MM-DD."""
return q("SELECT id, day, time FROM slots WHERE day = ? AND customer_id IS NULL ORDER BY time", (day,))
@mcp.tool
async def book_slot(slot_id: int, customer_id: int) -> str:
"""Book a free slot for a customer. Fails if the slot is already taken."""
n = w("UPDATE slots SET customer_id = ? WHERE id = ? AND customer_id IS NULL", (customer_id, slot_id))
if n == 0:
return f"Slot {slot_id} is already taken or does not exist."
row = q("SELECT day, time FROM slots WHERE id = ?", (slot_id,))[0]
return f"Booked slot {slot_id} ({row['day']} {row['time']}) for customer {customer_id}."
@mcp.tool
async def open_invoices(customer_id: int) -> list[dict]:
"""List a customer's invoices with their amounts and status."""
return q("SELECT id, amount, status FROM invoices WHERE customer_id = ?", (customer_id,))
@mcp.tool
async def issue_refund(invoice_id: int, ctx: Context) -> str | InputRequiredResult:
"""Refund an invoice. A human must approve the amount and give a reason."""
inv = q("SELECT id, customer_id, amount, status FROM invoices WHERE id = ?", (invoice_id,))
if not inv:
return f"No invoice {invoice_id}."
inv = inv[0]
responses = ctx.input_responses
if responses is None:
params = ElicitRequestFormParams(
message=(
f"Refund invoice {inv['id']} for customer {inv['customer_id']}? "
f"It is {inv['status']}, amount {inv['amount']:.2f}. "
"Confirm the amount to refund and give a reason."
),
requested_schema={
"type": "object",
"properties": {
"amount": {"type": "number", "description": "Amount to refund"},
"reason": {"type": "string", "description": "Why this refund is being issued"},
},
"required": ["amount", "reason"],
},
)
return InputRequiredResult(
result_type="input_required",
input_requests={"approve_refund": ElicitRequest(method="elicitation/create", params=params)},
)
answer = responses["approve_refund"]
if answer.action != "accept":
return f"Refund declined - invoice {inv['id']} unchanged."
w("UPDATE invoices SET status = 'refunded' WHERE id = ?", (inv["id"],))
return f"Refunded {answer.content['amount']:.2f} on invoice {inv['id']}. Reason: {answer.content['reason']}"
if __name__ == "__main__":
mcp.run(transport="http", host="127.0.0.1", port=8770, stateless_http=True)
Los docstrings no son comentarios. Se convierten en las descripciones que el modelo lee para elegir, así que escríbelos como instrucciones para alguien que no puede ver más que esa línea. Los dejamos en inglés porque son el texto que el modelo consume.
book_slot rechaza un hueco ocupado en SQL en vez de comprobarlo antes — WHERE customer_id IS NULL significa que dos agentes compitiendo por el mismo hueco no pueden ganar ambos.
cd ~/mcp-tools && ./.venv/bin/python office_tools.py
Paso 4 — La sesión que no debería existir
Déjalo corriendo. En una segunda sesión, pregúntale qué herramientas tiene:
curl -s -X POST http://127.0.0.1:8770/mcp -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'
La primera vez que ejecutamos esto, sin stateless_http=True:
{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"Bad Request: Missing session ID"}}
La revisión 2026-07-28 eliminó las sesiones del protocolo — sin handshake, sin ID de sesión,
nada que fije un cliente a una instancia. Todo lo escrito sobre el cambio describe bien el lado
cliente: no queda nada que fijar.
FastMCP 4.0.2 sigue usando el transporte con estado por defecto. Optas por el nuevo
comportamiento con stateless_http=True, y hasta que lo hagas, la primera petición contra tu
servidor recién creado falla citando un concepto que la especificación borró hace cinco semanas.
run_http_async acepta tanto stateless_http como stateless; el docstring dice que el
segundo es "Alias for stateless_http for CLI consistency". Un interruptor, dos nombres, ambos
con valor por defecto None.
La línea de arranque es donde lo compruebas:
Con la bandera, la misma petición devuelve el catálogo — sin handshake, sin sesión:
El esquema de issue_refund contiene solo invoice_id — el parámetro ctx: Context no está,
porque FastMCP lo inyecta y el modelo no puede fijarlo. Eso es lo que hace que una confirmación
obligatoria sea obligatoria.
Paso 5 — Un agente que descubre sus herramientas
import asyncio, os, sys
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
base_url="http://127.0.0.1:4000/v1",
api_key=os.environ["GATEWAY_KEY"],
model="qwen-mid",
temperature=0,
)
async def main():
async with MCPAdapter("http://127.0.0.1:8770/mcp") as adapter:
tools = await adapter.list_tools()
print("discovered:", [t.name for t in tools])
agent = create_agent(model, tools)
out = await agent.ainvoke({"messages": [{"role": "user", "content": sys.argv[1]}]})
print("\n--- answer ---")
print(out["messages"][-1].content)
asyncio.run(main())
Todos los ejemplos publicados de MCP apuntan el modelo a Anthropic, OpenAI o Gemini. Este apunta a tu propia gateway LiteLLM, que reenvía a WEC Inference — las herramientas son locales, el modelo es tuyo, nada sale de la caja.
Guarda la clave en un archivo y no en la línea de comandos, donde acabaría en tu historial de shell y en cada captura de pantalla:
cd ~/mcp-tools && printf 'GATEWAY_KEY=%s\n' "$KEY" > .env && chmod 600 .env
cd ~/mcp-tools && set -a && source .env && set +a && ./.venv/bin/python agent.py "Who is Maria and what invoices does she have?"
Esa respuesta necesitó dos llamadas a herramientas en secuencia: find_customer("Maria")
para obtener su id, y luego open_invoices(1) usándolo. Nada encaminó eso — el modelo leyó
cinco descripciones y dedujo el orden.
Ahora una que escribe:
cd ~/mcp-tools && ./.venv/bin/python agent.py "Book Maria into a free slot on 2026-09-04"
Tres llamadas esta vez. Y como una reserva es un cambio real, compruébalo donde el agente no llega:
cd ~/mcp-tools && ./.venv/bin/python -c "
import sqlite3
for r in sqlite3.connect('office.db').execute('select s.id,s.day,s.time,coalesce(c.name,\"free\") from slots s left join customers c on c.id=s.customer_id order by s.day,s.time'): print(r)"
Puedes levantar un servidor MCP sobre un almacén de datos real, apuntarle un agente, y hacer que encadene herramientas descubiertas hasta producir un cambio que verificas en la base de datos — con el modelo en tu propio gateway y no en la de un proveedor.
Paso 6 — La herramienta que se niega a terminar sola
La pausa de aprobación de la Parte 1 era algo que tú pusiste en el grafo. Esta viene de la herramienta, y el agente no tiene forma de saltársela.
import asyncio, json, os
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
model = ChatOpenAI(base_url="http://127.0.0.1:4000/v1",
api_key=os.environ["GATEWAY_KEY"], model="qwen-mid", temperature=0)
def ask_human(req):
print(f"\n>>> {req['message']}")
props = (req.get("requested_schema") or {}).get("properties", {})
content = {}
for name, spec in props.items():
raw = input(f" {name} ({spec.get('type','string')}): ").strip()
content[name] = float(raw) if spec.get("type") == "number" else raw
return {"action": "accept", "content": content}
async def main():
async with MCPAdapter("http://127.0.0.1:8770/mcp") as adapter:
tools = await adapter.list_tools()
agent = create_agent(model, tools, checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "refund-1"}}
out = await agent.ainvoke(
{"messages": [{"role": "user", "content": "Refund Maria's open invoice."}]}, cfg)
if "__interrupt__" not in out:
print("NO INTERRUPT:", out["messages"][-1].content); return
payload = out["__interrupt__"][0].value
print("=== INTERRUPT ===")
print(json.dumps(payload, indent=2))
responses = {r["key"]: ask_human(r) for r in payload["requests"]}
print("\n=== RESUMING ===")
out = await agent.ainvoke(Command(resume={"responses": responses}), cfg)
print(out["messages"][-1].content)
asyncio.run(main())
El checkpointer y el thread_id no son opcionales — una ejecución interrumpida necesita algún
sitio donde esperar. El mismo requisito que la pausa de aprobación de la Parte 1.
Lee el orden de los hechos. El modelo resolvió "la factura abierta de Maria" como la factura 2 por sí solo — ella tiene dos, y eligió la abierta. Luego se detuvo. El importe y el motivo los tecleó un humano, y solo entonces ocurrió el reembolso.
La carga es {"type": "mcp_elicitation", "tool_name": "issue_refund", "requests": [...]}. type
es el discriminador, de modo que un grafo con varios tipos de interrupt puede saber cuál vino de
MCP. Cada petición lleva una key, un message para un humano, un mode de form o url, y
para un formulario el requested_schema que la respuesta debe satisfacer — aquí dos campos
obligatorios, uno de ellos numérico. Reanudas con una respuesta por clave.
Y verifica, otra vez fuera del agente:
Puedes hacer que una herramienta se niegue a terminar sin un humano — devolviendo un formulario tipado que el modelo no puede rellenar por su cuenta, pausando la ejecución como un interrupt de LangGraph, y reanudándola con una respuesta que una persona tecleó de verdad.
ctx.elicit es la API vieja, y su error aterriza en el contexto del modeloTodos los ejemplos de elicitación de FastMCP que encontrarás llaman a
await ctx.elicit(message, response_type) dentro de la herramienta. Ese es el mecanismo de la
era del handshake: bloquea a mitad de ejecución y habla por el canal de retorno de la sesión —
el canal de retorno que el transporte sin estado no tiene.
Nuestro primer intento lo usó. El agente no se pausó — pero no porque no se lanzara nada.
FastMCP lanzó su error de era exactamente como está documentado, LangChain lo convirtió en un
ToolMessage con status="error" (deliberadamente, "en lugar de terminar la ejecución"), y el
modelo leyó ese error y lo narró como progreso:
Sin interrupt, código de salida 0, y la factura sigue open. No se reembolsó nada y nada está
esperando a un humano — pero al usuario se le dijo que hay una aprobación pendiente.
Con el protocolo moderno una herramienta devuelve un InputRequiredResult describiendo lo
que necesita, y sale. El cliente responde y reintenta la llamada completa con la respuesta
adjunta. Cada ronda es una petición completa, y por eso sobrevive al balanceo de carga y a los
redespliegues — y por eso el cuerpo de la herramienta debe escribirse para ser reentrado, no
reanudado. El nuestro lo hace con if ctx.input_responses is None.
La documentación de FastMCP es explícita: ctx.elicit solo funciona en conexiones
≤ 2025-11-25, y un servidor multi-era debería ramificar según
ctx.request_context.protocol_version. También promete que el desajuste "lanza un error de era
claro en lugar de fallar de forma oscura" — y así es. El error simplemente aterriza en el
contexto del modelo y no en el tuyo.
El post de LangChain lee el interrupt como paused["__interrupt__"][0].value.requests[0] —
acceso por atributo. MCPElicitationInterrupt es un TypedDict (ver
langchain/mcp/elicitation.py en 1.4.0), así que en tiempo de ejecución es un dict y
.requests no resuelve. Usa value["requests"] — como hace el propio fragmento unas líneas
después, leyendo question["key"].
Paso 7 — La caché, y por qué aquí no hace nada
Cada ejecución empieza descubriendo herramientas, lo que es un viaje de ida y vuelta antes de que el modelo vea nada. La nueva especificación lo hace cacheable:
import asyncio, time
from fastmcp import Client
from langchain.mcp import MCPAdapter
async def main():
client = Client("http://127.0.0.1:8770/mcp", cache=True)
async with MCPAdapter(client) as adapter:
for i in (1, 2):
t = time.perf_counter()
tools = await adapter.list_tools(cache_mode="use")
print(f"discovery {i}: {len(tools)} tools in {(time.perf_counter()-t)*1000:.1f} ms")
asyncio.run(main())
discovery 1: 5 tools in 14.6 ms
discovery 2: 5 tools in 12.5 ms
Sin efecto — y ese es el comportamiento correcto, no una caché rota.
cache=True es necesario, no suficienteLa caché del cliente "respeta las pistas ttlMs y cacheScope que el servidor adjunta a cada
respuesta" y solo funciona contra servidores de era moderna que las anuncien. El nuestro no
anuncia ninguna, así que ambas llamadas van a la red.
Fíjate además en que cache=True va en el fastmcp.Client, no en MCPAdapter — el
adaptador acepta exactamente un argumento. Y la caché pertenece al cliente, así que un cliente
por llamante evita que los catálogos se crucen entre inquilinos.
Por loopback no había nada que ahorrar de todos modos: 12–15 ms es el viaje de ida y vuelta. La caché es para servidores remotos con latencia real y un TTL que respetar.
Resolución de problemas — los errores que esta ejecución produjo de verdad
ResolutionImpossible al instalar mcp[cli]
El extra cli se eliminó en mcp 2.x. Instala "langchain[mcp]" en su lugar; arrastra un
mcp y un fastmcp compatibles.
Bad Request: Missing session ID
Tu servidor está usando el transporte con estado. Añade stateless_http=True a mcp.run(...).
KeyError: 'GATEWAY_KEY'
source .env define variables de shell, no variables de entorno — el proceso hijo de Python
nunca las ve. Usa set -a && source .env && set +a. Y export no cruza entre sesiones SSH, así
que una segunda terminal necesita su propio source.
address already in use tras reiniciar el servidor
La instancia anterior sigue ocupando el puerto — un reinicio fallido deja vivo el proceso
antiguo, así que el arreglo que acabas de hacer parece no funcionar. pkill -f office_tools.py,
y luego comprueba con ss -tlnp | grep 8770.
El agente nunca se pausa en una herramienta destructiva
La herramienta está llamando a ctx.elicit en lugar de devolver InputRequiredResult. En una
conexión sin estado no hay canal de retorno, y la llamada no lanza el error donde lo esperarías.
Qué te compró esto y qué no
Hecho: un servidor MCP sobre un almacén de datos real, un agente que descubre herramientas en lugar de estar cableado a ellas y las encadena sin encaminamiento, dos cambios que verificaste en la base de datos en vez de creerte al modelo, y un reembolso que no puede ocurrir sin que un humano aporte la cifra.
No hecho:
- Sin autenticación en el servidor MCP. Está enlazado a
127.0.0.1y cualquiera en la caja puede llamar a todas las herramientas, incluidaissue_refund. FastMCP soporta tokens bearer y OAuth 2.1; no configuramos ninguno. - Las herramientas confían por completo en quien llama.
book_slotreservará para cualquiercustomer_idque reciba. No hay noción de quién pregunta — el mismo hueco que la Parte 4 de la serie de endurecimiento encontró entre un plano de control y un plano de datos. - Un solo servidor.
ClientGroupconecta varios a la vez y prefija los nombres de herramienta por servidor; nosotros usamos un único objetivo. - SQLite, un solo proceso. Bien para un agente; la guarda a nivel de fila de
book_slothace más trabajo del que parece bajo concurrencia real. langchain.mcpestá en beta y avisa en cada importación.
Qué sigue
Acotar las herramientas por agente. Ahora mismo un solo agente tiene las cinco, así que nada
impide que el modelo elija issue_refund cuando se le pidió reservar un hueco — la pausa humana
es la única guarda. El supervisor de la Parte 2 ya reparte entre un agente de agenda y uno de
facturación; dale a cada uno solo su porción de adapter.list_tools() y el de agenda no podrá
emitir un reembolso porque nunca se le dijo que la herramienta existe. Eso es una garantía más
fuerte que pedírselo amablemente a un modelo.
Luego autenticación, para que una llamada a herramienta lleve una identidad en vez de confiar en quien alcance el puerto — la misma pregunta que la Parte 4 de la serie de endurecimiento hizo sobre la propia API del gateway, una capa más arriba.
Lecturas adicionales
- MCP en LangChain — inicio rápido, conexiones, auth, elicitación
- Migrar desde
langchain-mcp-adapters - Cliente FastMCP — transportes, auth, caché de respuestas
- Especificación MCP
2026-07-28 - Nuestro análisis de la revisión de la especificación