Requests HTTP

RequestsGetTool y RequestsPostTool

Las herramientas HTTP de LangChain proporcionan una interfaz directa para que los modelos de lenguaje puedan interactuar con APIs REST externas. Estas herramientas transforman las capacidades de los LLMs, permitiéndoles acceder a datos en tiempo real y realizar operaciones sobre servicios web de forma autónoma.

La biblioteca langchain_community incluye herramientas especializadas para los métodos HTTP más comunes, siendo RequestsGetTool y RequestsPostTool las fundamentales para la mayoría de integraciones con APIs RESTful.

Configuración básica de RequestsGetTool

La RequestsGetTool permite a los LLMs realizar peticiones GET para obtener información de APIs externas. Su configuración requiere un wrapper que gestione las respuestas HTTP:

from langchain_community.tools import RequestsGetTool
from langchain_community.utilities import TextRequestsWrapper
from langchain_openai import ChatOpenAI
import os

# Configurar el wrapper para manejar respuestas HTTP
requests_wrapper = TextRequestsWrapper()

# Crear la herramienta GET
get_tool = RequestsGetTool(
    requests_wrapper=requests_wrapper,
    allow_dangerous_requests=True  # Necesario para habilitar la herramienta
)

# Configurar el modelo
llm = ChatOpenAI(model="gpt-4", temperature=0)

El parámetro allow_dangerous_requests debe establecerse como True para habilitar la funcionalidad, ya que LangChain considera estas herramientas potencialmente peligrosas por su capacidad de realizar peticiones HTTP arbitrarias.

Uso práctico de RequestsGetTool

Para demostrar el funcionamiento, utilizaremos la API pública JSONPlaceholder que proporciona datos de prueba sin requerir autenticación:

from langchain.schema import HumanMessage
from langchain_core.tools import tool

# Ejemplo de uso directo de la herramienta
response = get_tool.invoke({
    "url": "https://jsonplaceholder.typicode.com/posts/1"
})

print(response)
# Salida: Contenido JSON del post con ID 1

Para integrar la herramienta con un agente conversacional, podemos crear un sistema que responda preguntas utilizando datos de la API:

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate

# Crear un prompt que incluya la herramienta
prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content="Eres un asistente que puede consultar APIs para obtener información actualizada."),
    HumanMessage(content="{question}")
])

# Vincular la herramienta al modelo
llm_with_tools = llm.bind_tools([get_tool])

# Crear una cadena LCEL
chain = prompt | llm_with_tools

# Ejemplo de consulta
result = chain.invoke({
    "question": "Obtén información sobre el post con ID 2 de JSONPlaceholder"
})

print(result.content)

Configuración de RequestsPostTool

La RequestsPostTool permite realizar peticiones POST para enviar datos a APIs externas. Su configuración es similar a la herramienta GET, pero incluye capacidades para manejar cuerpos de petición:

from langchain_community.tools import RequestsPostTool

# Crear la herramienta POST
post_tool = RequestsPostTool(
    requests_wrapper=requests_wrapper,
    allow_dangerous_requests=True
)

# Ejemplo de uso para crear un nuevo recurso
post_response = post_tool.invoke({
    "url": "https://jsonplaceholder.typicode.com/posts",
    "data": {
        "title": "Mi nuevo post",
        "body": "Contenido del post creado desde LangChain",
        "userId": 1
    }
})

print(post_response)

Manejo de headers y autenticación

El TextRequestsWrapper permite configurar headers personalizados y autenticación básica para APIs que lo requieran:

# Configurar wrapper con headers personalizados
requests_wrapper = TextRequestsWrapper(
    headers={
        "User-Agent": "LangChain-App/1.0",
        "Content-Type": "application/json"
    }
)

# Para APIs que requieren autenticación
api_key = os.getenv("API_KEY")  # Nunca hardcodear API keys
authenticated_wrapper = TextRequestsWrapper(
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
)

# Crear herramientas con autenticación
auth_get_tool = RequestsGetTool(
    requests_wrapper=authenticated_wrapper,
    allow_dangerous_requests=True
)

Integración con múltiples herramientas HTTP

LangChain proporciona herramientas adicionales para operaciones HTTP completas. Podemos combinar múltiples herramientas en un mismo agente:

from langchain_community.tools import RequestsPatchTool, RequestsDeleteTool

# Crear conjunto completo de herramientas HTTP
patch_tool = RequestsPatchTool(
    requests_wrapper=requests_wrapper,
    allow_dangerous_requests=True
)

delete_tool = RequestsDeleteTool(
    requests_wrapper=requests_wrapper,
    allow_dangerous_requests=True
)

# Lista de herramientas disponibles
http_tools = [get_tool, post_tool, patch_tool, delete_tool]

# Vincular todas las herramientas al modelo
llm_with_http_tools = llm.bind_tools(http_tools)

Manejo de respuestas y errores

El TextRequestsWrapper procesa automáticamente las respuestas HTTP y las convierte en texto legible para el LLM. Para un control más granular sobre las respuestas:

# Configurar wrapper con límites de respuesta
limited_wrapper = TextRequestsWrapper(
    headers={"User-Agent": "LangChain-Bot"},
    # Limitar el tamaño de respuesta para evitar tokens excesivos
)

# Crear herramienta con manejo de errores
robust_get_tool = RequestsGetTool(
    requests_wrapper=limited_wrapper,
    allow_dangerous_requests=True
)

# Ejemplo con manejo de errores en la cadena
try:
    response = robust_get_tool.invoke({
        "url": "https://api.ejemplo.com/datos"
    })
    print(f"Respuesta exitosa: {response}")
except Exception as e:
    print(f"Error en la petición: {e}")

Estas herramientas HTTP forman la base para crear agentes inteligentes capaces de interactuar con cualquier API REST, desde servicios de datos públicos hasta sistemas empresariales complejos. La clave está en configurar correctamente la autenticación y el manejo de respuestas según los requisitos específicos de cada API.

Integración con APIs RESTful externas

La integración con APIs RESTful representa uno de los casos de uso más valiosos de las herramientas HTTP en LangChain, permitiendo que los modelos de lenguaje accedan a datos actualizados y servicios especializados en tiempo real. Esta capacidad transforma los LLMs de sistemas estáticos a agentes dinámicos capaces de interactuar con el ecosistema digital actual.

Ejemplo práctico: API del clima OpenWeatherMap

Para demostrar una integración completa, implementaremos un sistema que consulte información meteorológica utilizando la API de OpenWeatherMap:

from langchain_community.tools import RequestsGetTool
from langchain_community.utilities import TextRequestsWrapper
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage
import os

# Configurar variables de entorno de forma segura
OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY")

# Crear wrapper con autenticación
weather_wrapper = TextRequestsWrapper(
    headers={
        "User-Agent": "LangChain-Weather-Bot/1.0"
    }
)

# Configurar herramienta específica para clima
weather_tool = RequestsGetTool(
    requests_wrapper=weather_wrapper,
    allow_dangerous_requests=True
)

# Configurar el modelo
llm = ChatOpenAI(model="gpt-4", temperature=0)

Creación de un agente meteorológico especializado

El siguiente ejemplo muestra cómo crear un agente especializado que puede interpretar consultas de clima y realizar las peticiones API apropiadas:

# Prompt especializado para consultas meteorológicas
weather_prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content="""
    Eres un asistente meteorológico especializado. Puedes consultar información del clima actual
    usando la API de OpenWeatherMap. Para obtener datos del clima, usa esta URL:
    https://api.openweathermap.org/data/2.5/weather?q={ciudad}&appid={api_key}&units=metric&lang=es
    
    Interpreta los datos JSON y proporciona respuestas claras y útiles sobre el clima.
    """),
    HumanMessage(content="{consulta}")
])

# Vincular herramienta al modelo
weather_llm = llm.bind_tools([weather_tool])

# Crear cadena LCEL
weather_chain = weather_prompt | weather_llm

# Función auxiliar para construir URLs
def build_weather_url(city: str) -> str:
    return f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={OPENWEATHER_API_KEY}&units=metric&lang=es"

# Ejemplo de uso
response = weather_chain.invoke({
    "consulta": f"¿Cómo está el clima en Madrid? Consulta: {build_weather_url('Madrid')}"
})

Integración con APIs de comercio electrónico

Las APIs de e-commerce como la de productos falsos de FakeStore API permiten crear asistentes de compras inteligentes:

# Configurar herramientas para API de productos
store_wrapper = TextRequestsWrapper(
    headers={
        "Content-Type": "application/json",
        "User-Agent": "LangChain-Store-Assistant/1.0"
    }
)

store_get_tool = RequestsGetTool(
    requests_wrapper=store_wrapper,
    allow_dangerous_requests=True
)

# Prompt para asistente de compras
shopping_prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content="""
    Eres un asistente de compras que puede consultar productos usando la FakeStore API.
    URLs disponibles:
    - Todos los productos: https://fakestoreapi.com/products
    - Producto específico: https://fakestoreapi.com/products/{id}
    - Categorías: https://fakestoreapi.com/products/categories
    - Por categoría: https://fakestoreapi.com/products/category/{categoria}
    
    Analiza los datos JSON y proporciona recomendaciones útiles.
    """),
    HumanMessage(content="{consulta}")
])

# Crear asistente de compras
shopping_llm = llm.bind_tools([store_get_tool])
shopping_chain = shopping_prompt | shopping_llm

Manejo de APIs con paginación

Muchas APIs RESTful implementan paginación para manejar grandes conjuntos de datos. Aquí se muestra cómo gestionar este patrón:

from langchain_core.tools import tool

@tool
def fetch_paginated_data(base_url: str, page: int = 1, limit: int = 10) -> str:
    """
    Herramienta personalizada para manejar APIs con paginación.
    
    Args:
        base_url: URL base de la API
        page: Número de página a consultar
        limit: Cantidad de elementos por página
    """
    paginated_url = f"{base_url}?page={page}&limit={limit}"
    
    # Usar la herramienta GET existente
    response = store_get_tool.invoke({"url": paginated_url})
    return response

# Integrar herramienta de paginación
paginated_tools = [store_get_tool, fetch_paginated_data]
paginated_llm = llm.bind_tools(paginated_tools)

Gestión de autenticación OAuth 2.0

Para APIs empresariales que requieren OAuth 2.0, podemos configurar el wrapper con tokens de acceso:

# Configuración para APIs con OAuth 2.0
oauth_token = os.getenv("OAUTH_ACCESS_TOKEN")

oauth_wrapper = TextRequestsWrapper(
    headers={
        "Authorization": f"Bearer {oauth_token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
)

# Herramientas para API empresarial
enterprise_get_tool = RequestsGetTool(
    requests_wrapper=oauth_wrapper,
    allow_dangerous_requests=True
)

enterprise_post_tool = RequestsPostTool(
    requests_wrapper=oauth_wrapper,
    allow_dangerous_requests=True
)

Implementación de rate limiting y reintentos

Las APIs externas suelen implementar límites de velocidad. Podemos crear un wrapper personalizado que maneje estos escenarios:

import time
from typing import Dict, Any

class RateLimitedWrapper(TextRequestsWrapper):
    def __init__(self, requests_per_minute: int = 60, **kwargs):
        super().__init__(**kwargs)
        self.requests_per_minute = requests_per_minute
        self.last_request_time = 0
        self.min_interval = 60.0 / requests_per_minute
    
    def get(self, url: str, **kwargs) -> str:
        # Implementar rate limiting
        current_time = time.time()
        time_since_last = current_time - self.last_request_time
        
        if time_since_last < self.min_interval:
            time.sleep(self.min_interval - time_since_last)
        
        self.last_request_time = time.time()
        return super().get(url, **kwargs)

# Usar wrapper con rate limiting
rate_limited_wrapper = RateLimitedWrapper(
    requests_per_minute=30,  # Máximo 30 peticiones por minuto
    headers={"User-Agent": "LangChain-RateLimited/1.0"}
)

rate_limited_tool = RequestsGetTool(
    requests_wrapper=rate_limited_wrapper,
    allow_dangerous_requests=True
)

Integración con APIs de datos financieros

Un ejemplo avanzado utilizando APIs financieras para crear un asistente de inversiones:

# Configuración para API financiera (ejemplo con Alpha Vantage)
finance_api_key = os.getenv("ALPHA_VANTAGE_API_KEY")

finance_wrapper = TextRequestsWrapper(
    headers={
        "User-Agent": "LangChain-Finance-Bot/1.0"
    }
)

finance_tool = RequestsGetTool(
    requests_wrapper=finance_wrapper,
    allow_dangerous_requests=True
)

# Prompt especializado en finanzas
finance_prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content=f"""
    Eres un asistente financiero que puede consultar datos de mercado en tiempo real.
    Usa la API de Alpha Vantage con esta estructura:
    https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={{symbol}}&apikey={finance_api_key}
    
    Interpreta los datos financieros y proporciona análisis claros sobre precios de acciones.
    """),
    HumanMessage(content="{consulta}")
])

# Crear asistente financiero
finance_llm = llm.bind_tools([finance_tool])
finance_chain = finance_prompt | finance_llm

Consideraciones de seguridad en producción

Al implementar integraciones con APIs externas en entornos de producción, es fundamental seguir estas prácticas de seguridad:

Gestión segura de credenciales:

import os
from pathlib import Path

# Cargar configuración desde archivo .env
def load_api_config():
    config = {
        "weather_api_key": os.getenv("OPENWEATHER_API_KEY"),
        "finance_api_key": os.getenv("ALPHA_VANTAGE_API_KEY"),
        "oauth_token": os.getenv("OAUTH_ACCESS_TOKEN")
    }
    
    # Validar que todas las claves estén presentes
    missing_keys = [k for k, v in config.items() if not v]
    if missing_keys:
        raise ValueError(f"Faltan las siguientes variables de entorno: {missing_keys}")
    
    return config

# Usar configuración segura
try:
    api_config = load_api_config()
except ValueError as e:
    print(f"Error de configuración: {e}")

Validación de URLs y filtrado de dominios:

from urllib.parse import urlparse

def validate_api_url(url: str, allowed_domains: list) -> bool:
    """Validar que la URL pertenezca a dominios permitidos"""
    parsed = urlparse(url)
    return parsed.netloc in allowed_domains

# Lista de dominios permitidos
ALLOWED_DOMAINS = [
    "api.openweathermap.org",
    "fakestoreapi.com",
    "jsonplaceholder.typicode.com"
]

# Wrapper con validación de seguridad
class SecureRequestsWrapper(TextRequestsWrapper):
    def __init__(self, allowed_domains: list, **kwargs):
        super().__init__(**kwargs)
        self.allowed_domains = allowed_domains
    
    def get(self, url: str, **kwargs) -> str:
        if not validate_api_url(url, self.allowed_domains):
            raise ValueError(f"Dominio no permitido: {urlparse(url).netloc}")
        return super().get(url, **kwargs)

Esta aproximación de integración con APIs RESTful permite crear sistemas LangChain robustos y seguros que pueden interactuar con prácticamente cualquier servicio web moderno, desde APIs públicas simples hasta sistemas empresariales complejos con autenticación avanzada.