@wrap_tool_call: Personalizar Ejecución de Tools

Descripción de la cápsula

Así como @wrap_model_call envuelve las llamadas al modelo, @wrap_tool_call envuelve la ejecución de tools. Cada vez que el agente decide llamar una tool, tu middleware intercepta esa ejecución antes de que ocurra y después de que termina. Esto te permite agregar retry logic, custom error handling, logging detallado, timing, y cualquier lógica que necesites alrededor de tus tools — sin modificar el código de las tools mismas.

En la cápsula anterior aprendiste a interceptar llamadas al modelo con @wrap_model_call. Ahora completas el otro lado: las tools. Con ambos middleware combinados, tienes visibilidad y control total sobre cada operación que el agente ejecuta. Al terminar esta cápsula, sabrás construir agentes que manejan errores de tools de forma inteligente, miden el rendimiento de cada tool, y registran logs detallados de cada ejecución.

El patrón handler

@wrap_tool_call recibe tres parámetros:

1. tool_call — Un diccionario con name (nombre de la tool), args (argumentos), e id (identificador único del call).
2. config — La configuración de ejecución (misma que en @wrap_model_call).
3. call_next — La función que ejecuta la tool real. Siempre debes llamarla (a menos que quieras bloquear la ejecución).

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    return f"El clima en {city} es soleado, 22°C"

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

def wrap_tool_call(tool_call, config, call_next):
    """Intercepta cada ejecución de tool."""
    print(f"[TOOL] Ejecutando: {tool_call['name']}({tool_call['args']})")

    result = call_next(tool_call, config)

    print(f"[TOOL] Resultado: {result}")
    return result

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[get_weather, calculator],
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({
    "messages": [("user", "¿Qué clima hace en Madrid y cuánto es 15 * 8?")]
})
print(f"\nRespuesta: {result['messages'][-1].content}")

# Output esperado:
# [TOOL] Ejecutando: get_weather({'city': 'Madrid'})
# [TOOL] Resultado: El clima en Madrid es soleado, 22°C
# [TOOL] Ejecutando: calculator({'expression': '15 * 8'})
# [TOOL] Resultado: 120
#
# Respuesta: El clima en Madrid es soleado, 22°C. Y 15 × 8 = 120.

El middleware se ejecutó una vez por cada tool call. Cuando el modelo pide dos tools en paralelo, tu middleware intercepta ambas individualmente.

El id dentro de tool_call se usa internamente para emparejar cada ToolMessage con su tool call correspondiente — no necesitas manipularlo.

Retry logic: reintentar tools que fallan

Las tools pueden fallar — APIs caídas, timeouts, errores de red. @wrap_tool_call permite agregar reintentos automáticos:

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.agents import create_agent
from langchain_core.tools import tool

attempt_counter = 0

@tool
def flaky_api(query: str) -> str:
    """Consulta una API externa que a veces falla."""
    global attempt_counter
    attempt_counter += 1
    if attempt_counter < 3:
        raise ConnectionError(f"API timeout (intento {attempt_counter})")
    return f"Datos para '{query}': temperatura 22°C, humedad 65%"

def wrap_tool_call(tool_call, config, call_next):
    """Reintenta tools con backoff exponencial."""
    tool_name = tool_call["name"]
    max_retries = 3

    for attempt in range(max_retries):
        try:
            result = call_next(tool_call, config)
            if attempt > 0:
                print(f"[RETRY] {tool_name} exitoso en intento {attempt + 1}")
            return result
        except Exception as e:
            wait_time = 2 ** attempt
            if attempt < max_retries - 1:
                print(f"[RETRY] {tool_name} falló ({attempt + 1}/{max_retries}): {e}")
                time.sleep(wait_time)
            else:
                print(f"[RETRY] {tool_name} falló después de {max_retries} intentos")
                raise

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[flaky_api],
    wrap_tool_call=wrap_tool_call,
)

attempt_counter = 0
result = agent.invoke({"messages": [("user", "¿Qué temperatura hace?")]})
print(f"\nRespuesta: {result['messages'][-1].content}")

# Output esperado:
# [RETRY] flaky_api falló (1/3): API timeout (intento 1)
# [RETRY] flaky_api falló (2/3): API timeout (intento 2)
# [RETRY] flaky_api exitoso en intento 3
#
# Respuesta: La temperatura actual es de 22°C con una humedad del 65%.

El backoff exponencial (1s, 2s, 4s...) evita sobrecargar APIs con problemas.

Custom error handling: errores amigables

En vez de dejar que los errores rompan el agente, captura excepciones y retorna mensajes informativos:

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def get_stock_price(symbol: str) -> str:
    """Obtiene el precio actual de una acción."""
    prices = {"AAPL": "$178.50", "GOOGL": "$141.20"}
    if symbol not in prices:
        raise ValueError(f"Símbolo '{symbol}' no encontrado")
    return f"{symbol}: {prices[symbol]}"

@tool
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    """Obtiene el tipo de cambio entre divisas."""
    rates = {("USD", "MXN"): "17.15"}
    key = (from_currency.upper(), to_currency.upper())
    if key not in rates:
        raise ConnectionError("Servicio de divisas no disponible")
    return f"1 {from_currency} = {rates[key]} {to_currency}"

def wrap_tool_call(tool_call, config, call_next):
    """Maneja errores por tipo con mensajes útiles."""
    tool_name = tool_call["name"]
    try:
        return call_next(tool_call, config)
    except ValueError as e:
        print(f"[ERROR] {tool_name}: dato no encontrado — {e}")
        return f"Error en {tool_name}: {e}. Intenta con un valor diferente."
    except ConnectionError as e:
        print(f"[ERROR] {tool_name}: servicio caído — {e}")
        return f"Error en {tool_name}: servicio no disponible. Intenta más tarde."
    except Exception as e:
        print(f"[ERROR] {tool_name}: error inesperado — {type(e).__name__}: {e}")
        return f"Error inesperado en {tool_name}."

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[get_stock_price, get_exchange_rate],
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({
    "messages": [("user", "¿Cuánto cuesta la acción de TSLA en pesos mexicanos?")]
})
print(result["messages"][-1].content)

# Output esperado:
# [ERROR] get_stock_price: dato no encontrado — Símbolo 'TSLA' no encontrado
# No pude obtener el precio de TSLA. El símbolo no fue encontrado...

El modelo recibe un mensaje de error claro y lo integra en su respuesta. El agente sigue funcionando.

Tool-level logging y timing

Registra cada tool call con detalles completos y mide su duración:

from dotenv import load_dotenv
load_dotenv()

import time
import json
from datetime import datetime
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado para '{query}': Python es un lenguaje interpretado."

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

tool_log = []

def wrap_tool_call(tool_call, config, call_next):
    """Registra cada tool call con timestamp, duración, y status."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "tool": tool_call["name"],
        "args": tool_call["args"],
        "status": "pending",
    }
    start = time.time()

    try:
        result = call_next(tool_call, config)
        entry["status"] = "success"
        entry["result"] = str(result)[:200]
        return result
    except Exception as e:
        entry["status"] = "error"
        entry["error"] = f"{type(e).__name__}: {e}"
        raise
    finally:
        entry["duration_ms"] = round((time.time() - start) * 1000, 1)
        tool_log.append(entry)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search, calculator],
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({
    "messages": [("user", "¿Qué es Python y cuánto es 2**10?")]
})
print(f"Respuesta: {result['messages'][-1].content}\n")

print("=== Tool Log ===")
for entry in tool_log:
    print(f"  {entry['tool']} ({entry['duration_ms']}ms) → {entry['status']}")
    print(f"    Args: {entry['args']}")

# Output esperado:
# Respuesta: Python es un lenguaje interpretado. 2^10 = 1,024.
#
# === Tool Log ===
#   search (1.2ms) → success
#     Args: {'query': 'qué es Python'}
#   calculator (0.3ms) → success
#     Args: {'expression': '2**10'}

El bloque try/except/finally garantiza que cada tool call se registra, sea exitosa o no. En producción, enviarías este log a CloudWatch, Datadog, etc.

Combinar @wrap_model_call y @wrap_tool_call

Para observabilidad completa, usa ambos middleware juntos:

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado: Python 3.12 incluye mejoras en performance."

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

session_log = []

def wrap_model_call(messages, config, call_next):
    start = time.time()
    response = call_next(messages, config)
    elapsed = round(time.time() - start, 3)
    tool_info = "→ tool calls" if response.tool_calls else "→ respuesta final"
    session_log.append(f"MODEL ({elapsed}s) {len(messages)} msgs {tool_info}")
    return response

def wrap_tool_call(tool_call, config, call_next):
    start = time.time()
    try:
        result = call_next(tool_call, config)
        elapsed = round(time.time() - start, 3)
        session_log.append(f"TOOL  ({elapsed}s) {tool_call['name']}({tool_call['args']}) → OK")
        return result
    except Exception as e:
        elapsed = round(time.time() - start, 3)
        session_log.append(f"TOOL  ({elapsed}s) {tool_call['name']} → ERROR: {e}")
        raise

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search, calculator],
    wrap_model_call=wrap_model_call,
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({
    "messages": [("user", "¿Qué versión de Python es la más nueva y cuánto es 7**4?")]
})
print(f"Respuesta: {result['messages'][-1].content}\n")

print("=== Session Log ===")
for entry in session_log:
    print(f"  {entry}")

# Output esperado:
# Respuesta: La versión más reciente de Python es 3.12... 7^4 = 2,401.
#
# === Session Log ===
#   MODEL (0.8s) 1 msgs → tool calls
#   TOOL  (0.001s) search({'query': '...'}) → OK
#   TOOL  (0.001s) calculator({'expression': '7**4'}) → OK
#   MODEL (0.6s) 4 msgs → respuesta final

El log muestra la secuencia completa: modelo decide → tools ejecutan → modelo responde. Visibilidad total del pipeline.

Comparación: @wrap_model_call vs @wrap_tool_call

Ambos middleware son complementarios. Juntos cubren los dos tipos de operación del agente.

Troubleshooting

Problema 1: "TypeError: wrap_tool_call() takes 2 positional arguments but 3 were given"

Síntoma: Error al crear o ejecutar el agente. Causa: Tu función tiene la firma incorrecta. Solución: La firma debe ser exactamente (tool_call, config, call_next):

def wrap_tool_call(tool_call, config, call_next):
    return call_next(tool_call, config)

Problema 2: El retry loop nunca termina

Síntoma: El agente se queda colgado indefinidamente. Causa: Tu retry logic no tiene límite máximo. Solución: Siempre establece max_retries y un fallback:

def wrap_tool_call(tool_call, config, call_next):
    for attempt in range(3):
        try:
            return call_next(tool_call, config)
        except Exception:
            if attempt == 2:
                return f"Error: {tool_call['name']} no disponible"
            time.sleep(2 ** attempt)

Problema 3: Error al modificar tool_call args

Síntoma: Error al intentar modificar tool_call["args"] directamente. Causa: El diccionario puede ser inmutable. Solución: Crea un nuevo diccionario:

def wrap_tool_call(tool_call, config, call_next):
    modified = {**tool_call, "args": {**tool_call["args"], "limit": 10}}
    return call_next(modified, config)

Problema 4: El middleware no se ejecuta para una tool

Síntoma: El middleware funciona para algunas tools pero no para otras. Causa: @wrap_tool_call se ejecuta para todas las tools. Si parece que no, el modelo no está llamando esa tool. Solución: Verifica que el modelo llama la tool revisando su docstring. El modelo decide qué tool usar basándose en la descripción.

Ejercicios

Ejercicio 1: Logging básico de tools (Fácil)

Crea un @wrap_tool_call que imprima nombre, args, y resultado de cada tool. Prueba con dos tools.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado: {query} — Python fue creado en 1991."

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

def wrap_tool_call(tool_call, config, call_next):
    print(f"[TOOL] {tool_call['name']}({tool_call['args']})")
    result = call_next(tool_call, config)
    print(f"[TOOL] → {str(result)[:100]}")
    return result

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search, calculator],
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({"messages": [("user", "¿Cuándo se creó Python y cuánto es 3**5?")]})
print(f"\nRespuesta: {result['messages'][-1].content}")

# Output esperado:
# [TOOL] search({'query': 'cuándo se creó Python'})
# [TOOL] → Resultado: cuándo se creó Python — Python fue creado en 1991.
# [TOOL] calculator({'expression': '3**5'})
# [TOOL] → 243
#
# Respuesta: Python fue creado en 1991. 3^5 = 243.

Ejercicio 2: Timing por tool (Fácil)

Crea un middleware que mida cuánto tarda cada tool e imprima el tiempo en milisegundos.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def instant_lookup(key: str) -> str:
    """Busca un dato local."""
    return f"Dato: {key} = 42"

@tool
def slow_search(query: str) -> str:
    """Busca en un servicio externo lento."""
    time.sleep(0.5)
    return f"Resultado para '{query}': información detallada."

def wrap_tool_call(tool_call, config, call_next):
    start = time.time()
    result = call_next(tool_call, config)
    elapsed_ms = (time.time() - start) * 1000
    print(f"[TIMING] {tool_call['name']}: {elapsed_ms:.1f}ms")
    return result

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[instant_lookup, slow_search],
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({
    "messages": [("user", "Busca 'pi' localmente y busca info sobre calculus")]
})
print(f"\nRespuesta: {result['messages'][-1].content}")

# Output esperado:
# [TIMING] instant_lookup: 0.2ms
# [TIMING] slow_search: 502.3ms
#
# Respuesta: El dato de pi es 42. Sobre calculus: información detallada.

Ejercicio 3: Retry con backoff exponencial (Medio)

Crea una tool que falle las primeras 2 veces y un middleware con retry (3 intentos, backoff). Verifica que el tercer intento es exitoso.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.agents import create_agent
from langchain_core.tools import tool

call_counter = 0

@tool
def unreliable_api(query: str) -> str:
    """API que falla intermitentemente."""
    global call_counter
    call_counter += 1
    if call_counter < 3:
        raise ConnectionError(f"Timeout en intento {call_counter}")
    return f"Datos para '{query}': respuesta exitosa."

def wrap_tool_call(tool_call, config, call_next):
    max_retries = 3
    for attempt in range(max_retries):
        try:
            result = call_next(tool_call, config)
            if attempt > 0:
                print(f"[RETRY] OK en intento {attempt + 1}")
            return result
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"[RETRY] Intento {attempt + 1}/{max_retries} falló: {e}. Esperando {wait}s...")
                time.sleep(wait)
            else:
                return f"Error: {tool_call['name']} no disponible tras {max_retries} intentos."

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[unreliable_api],
    wrap_tool_call=wrap_tool_call,
)

call_counter = 0
result = agent.invoke({"messages": [("user", "Consulta la API sobre LangChain")]})
print(f"\nRespuesta: {result['messages'][-1].content}")

# Output esperado:
# [RETRY] Intento 1/3 falló: Timeout en intento 1. Esperando 1s...
# [RETRY] Intento 2/3 falló: Timeout en intento 2. Esperando 2s...
# [RETRY] OK en intento 3
#
# Respuesta: Sobre LangChain: respuesta exitosa.

Ejercicio 4: Bloquear tools peligrosas (Medio)

Crea un agente con tools read, write, delete. El middleware bloquea delete retornando un error amigable. Verifica que las otras funcionan.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def read_data(key: str) -> str:
    """Lee datos del sistema."""
    return f"Datos para '{key}': valor_123"

@tool
def write_data(key: str, value: str) -> str:
    """Escribe datos en el sistema."""
    return f"Escrito: {key} = {value}"

@tool
def delete_data(key: str) -> str:
    """Elimina datos del sistema."""
    return f"Eliminado: {key}"

BLOCKED = {"delete_data"}

def wrap_tool_call(tool_call, config, call_next):
    if tool_call["name"] in BLOCKED:
        return f"'{tool_call['name']}' bloqueada. Solo lectura y escritura permitidas."
    return call_next(tool_call, config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[read_data, write_data, delete_data],
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({
    "messages": [("user", "Lee 'config', escribe 'status'='activo', y elimina 'temp'")]
})
print(result["messages"][-1].content)

# Output esperado:
# He leído 'config' (valor_123), escrito 'status' = 'activo'.
# La eliminación de 'temp' está bloqueada — solo lectura y escritura están permitidas.

Ejercicio 5: Observabilidad completa — model + tools (Challenge)

Combina @wrap_model_call y @wrap_tool_call. El model middleware cuenta llamadas y tokens. El tool middleware registra timing y status. Imprime un dashboard al final.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado: {query} — dato relevante."

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

dashboard = {
    "model_calls": 0, "model_time": 0.0,
    "input_tokens": 0, "output_tokens": 0,
    "tool_calls": 0, "tool_time": 0.0, "tool_errors": 0,
}

def wrap_model_call(messages, config, call_next):
    start = time.time()
    response = call_next(messages, config)
    dashboard["model_calls"] += 1
    dashboard["model_time"] += time.time() - start
    usage = response.response_metadata.get("token_usage", {})
    dashboard["input_tokens"] += usage.get("prompt_tokens", 0)
    dashboard["output_tokens"] += usage.get("completion_tokens", 0)
    return response

def wrap_tool_call(tool_call, config, call_next):
    start = time.time()
    try:
        result = call_next(tool_call, config)
        dashboard["tool_calls"] += 1
        dashboard["tool_time"] += time.time() - start
        return result
    except Exception as e:
        dashboard["tool_errors"] += 1
        dashboard["tool_calls"] += 1
        return f"Error: {e}"

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search, calculator],
    wrap_model_call=wrap_model_call,
    wrap_tool_call=wrap_tool_call,
)

result = agent.invoke({"messages": [("user", "¿Qué es AI y cuánto es 256 * 4?")]})
print(f"Respuesta: {result['messages'][-1].content}\n")

total_cost = (
    (dashboard["input_tokens"] / 1000) * 0.00015
    + (dashboard["output_tokens"] / 1000) * 0.0006
)
print("=== DASHBOARD ===")
print(f"Model: {dashboard['model_calls']} calls | {dashboard['model_time']:.2f}s")
print(f"Tokens: {dashboard['input_tokens']} in / {dashboard['output_tokens']} out")
print(f"Cost: ${total_cost:.6f} USD")
print(f"Tools: {dashboard['tool_calls']} calls | {dashboard['tool_time']:.3f}s | {dashboard['tool_errors']} errors")

# Output esperado:
# Respuesta: AI (Inteligencia Artificial) es... 256 × 4 = 1,024.
#
# === DASHBOARD ===
# Model: 2 calls | 1.23s
# Tokens: ~200 in / ~60 out
# Cost: $0.000066 USD
# Tools: 2 calls | 0.002s | 0 errors

Resumen

En esta cápsula aprendiste:

• @wrap_tool_call recibe (tool_call, config, call_next) y envuelve la ejecución de cada tool
• El tool_call es un dict con name, args, e id
• Retry logic: reintentar tools con backoff exponencial para manejar APIs inestables
• Custom error handling: capturar excepciones por tipo y retornar mensajes útiles para el modelo
• Tool-level logging: registrar cada ejecución con timestamp, args, resultado, y status
• Tool-level timing: medir duración para identificar cuellos de botella
• Bloquear tools: impedir ejecución de tools peligrosas retornando mensajes de error
• Observabilidad completa: combinar @wrap_model_call + @wrap_tool_call para visibilidad total
• El fallback pattern (retry → mensaje de error) es preferible a propagar excepciones

Próxima cápsula: Dynamic models — aprenderás a usar middleware para seleccionar automáticamente qué modelo usar según la complejidad, optimizando costo y latencia.

Recursos adicionales

1. create_agent API Reference — Documentación completa de parámetros de middleware
2. LangGraph Agents — Tool Middleware — Guía oficial de middleware de tools
3. Custom Error Handling in Agents — How-to de error handling custom
4. Tool Calling — LangChain — Conceptos de tool calling
5. Exponential Backoff — Google Cloud — Referencia sobre backoff exponencial

Módulo 4 — LangChain & LangGraph: From Chains to Agents