Bucle de herramientas

Nivel
OpenAI
OpenAI
Actualizado: 22/11/2025

Qué es el bucle de herramientas

El bucle de herramientas es un patrón en el que un modelo de lenguaje invoca funciones de Python de forma iterativa, recibe sus resultados estructurados y decide qué hacer a continuación hasta construir una respuesta útil para el usuario. En lugar de limitarse a generar texto, el LLM se apoya en herramientas externas (base de datos, APIs HTTP, utilidades diversas) que amplían sus capacidades con datos en tiempo real.

Con este enfoque el modelo se comporta como un agente siguiendo el ciclo: lees las peticiones de herramientas que genera el modelo, ejecutas el código real y vas alimentando de nuevo al modelo con los resultados. En este ejemplo completas el recorrido desde funciones normales en Python hasta un bucle de herramientas que resuelve una tarea de negocio realista.

Definir tools

En este bloque defines las tools como funciones Python puras que encapsulan operaciones bien acotadas: acceso a la base de datos de clientes, consultas a APIs públicas de países, tiempo, divisas o festivos, y algunas utilidades ligeras como datos curiosos o chistes.

Cada función devuelve un diccionario con una clave ok que indica si la operación ha ido bien y datos adicionales en campos como rows, current, rates o holidays, lo que simplifica el razonamiento posterior del modelo.

import os
import json
import requests
import re
from typing import Optional, List, Dict, Any
from sqlmodel import SQLModel, Field, Session, create_engine, select
from sqlalchemy import text
from openai import OpenAI

client = OpenAI()

# ---------- DB y Customer ----------
class Customer(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str
    country: str
    currency: str
    phone: Optional[str] = None
    balance: float = 0.0

DB_URL = "sqlite:///./customers.db"

def init_db(force: bool = False):
    if force and DB_URL.startswith("sqlite:///"):
        path = DB_URL.replace("sqlite:///", "")
        if os.path.exists(path):
            os.remove(path)
    engine = create_engine(DB_URL, connect_args={"check_same_thread": False})
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        if not session.exec(select(Customer)).first():
            customers = [
                Customer(name="Alicia García", email="alicia.garcia@example.com", country="ES", currency="EUR", phone="+34-600-111-111", balance=1250.50),
                Customer(name="Bruno Silva", email="bruno.silva@example.com", country="PT", currency="EUR", phone="+351-912-222-222", balance=980.00),
                Customer(name="Carla Rossi", email="carla.rossi@example.com", country="IT", currency="EUR", phone="+39-333-333-333", balance=305.75),
                Customer(name="David Müller", email="david.mueller@example.com", country="DE", currency="EUR", phone="+49-170-444-444", balance=4200.00),
                Customer(name="Emma Johnson", email="emma.johnson@example.com", country="ES", currency="GBP", phone="+44-7700-555-555", balance=60.00),
            ]
            session.add_all(customers)
            session.commit()

# ---------- Seguridad para SQL ----------
_FORBIDDEN_RE = re.compile(
    r"\b(insert|update|delete|drop|alter|create|replace|truncate|pragma|attach|detach|merge|call|grant|revoke)\b",
    flags=re.IGNORECASE,
)

def _is_safe_select(sql: str) -> bool:
    if not isinstance(sql, str):
        return False
    if ";" in sql:
        return False
    s_no_comments = re.sub(r"^\s*(--.*\n|/\*.*?\*/\s*)", "", sql, flags=re.DOTALL).lstrip()
    if not s_no_comments.lower().startswith("select"):
        return False
    if _FORBIDDEN_RE.search(sql):
        return False
    return True

def query_sqlite(sql: str) -> Dict[str, Any]:
    if not _is_safe_select(sql):
        return {"ok": False, "error": "Solo se permiten SELECT simples."}
    engine = create_engine(DB_URL, connect_args={"check_same_thread": False})
    try:
        with Session(engine) as session:
            result = session.exec(text(sql))
            rows = result.mappings().all()
            return {"ok": True, "rows": [dict(r) for r in rows], "row_count": len(rows)}
    except Exception as e:
        return {"ok": False, "error": str(e)}

# ---------- Weather ----------
def get_weather(latitude: float, longitude: float) -> Dict[str, Any]:
    try:
        url = f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}&current_weather=true"
        r = requests.get(url, timeout=10)
        r.raise_for_status()
        data = r.json()
        return {"ok": True, "current": data.get("current_weather", {}), "raw": data}
    except Exception as e:
        return {"ok": False, "error": str(e)}

# ---------- Country info ----------
def get_country_info(query: str) -> Dict[str, Any]:
    try:
        if len(query) in (2,3):
            url = f"https://restcountries.com/v3.1/alpha/{query}"
        else:
            url = f"https://restcountries.com/v3.1/name/{requests.utils.quote(query)}"
        r = requests.get(url, timeout=10)
        r.raise_for_status()
        data = r.json()
        country = data[0] if isinstance(data, list) else data
        capital = (country.get("capital") or [None])[0]
        latlng = country.get("latlng")
        return {"ok": True, "name": country.get("name", {}).get("common"), "capital": capital, "latlng": latlng, "raw": country}
    except Exception as e:
        return {"ok": False, "error": str(e)}

# ---------- Currency ----------
def get_exchange_rate(base: str = "USD", symbols: Optional[List[str]] = None) -> Dict[str, Any]:
    try:
        url = f"https://open.er-api.com/v6/latest/{base.upper()}"
        r = requests.get(url, timeout=10)
        r.raise_for_status()
        data = r.json()
        if data.get("result") != "success":
            return {"ok": False, "error": data}
        rates = data.get("rates", {})
        if symbols:
            rates = {k: v for k, v in rates.items() if k.upper() in [s.upper() for s in symbols]}
        return {"ok": True, "base": base.upper(), "rates": rates, "raw": data}
    except Exception as e:
        return {"ok": False, "error": str(e)}

def convert_currency(amount: float, from_currency: str, to_currency: str) -> Dict[str, Any]:
    try:
        rates_resp = get_exchange_rate(base=from_currency, symbols=[to_currency])
        if not rates_resp.get("ok"):
            return {"ok": False, "error": rates_resp.get("error")}
        rate = rates_resp["rates"][to_currency.upper()]
        converted = amount * rate
        return {"ok": True, "from": from_currency.upper(), "to": to_currency.upper(), "amount": amount, "converted": converted, "rate": rate, "raw": rates_resp}
    except Exception as e:
        return {"ok": False, "error": str(e)}

# ---------- Misc Tools ----------
def get_cat_fact():
    try:
        r = requests.get("https://catfact.ninja/fact", timeout=8)
        r.raise_for_status()
        return {"ok": True, "raw": r.json()}
    except Exception as e:
        return {"ok": False, "error": str(e)}

def get_random_joke():
    try:
        r = requests.get("https://official-joke-api.appspot.com/jokes/random", timeout=8)
        r.raise_for_status()
        return {"ok": True, "raw": r.json()}
    except Exception as e:
        return {"ok": False, "error": str(e)}

def get_sun_times(latitude: float, longitude: float, date: str = "today"):
    try:
        r = requests.get("https://api.sunrise-sunset.org/json", params={"lat": latitude, "lng": longitude, "date": date, "formatted": 0}, timeout=8)
        r.raise_for_status()
        j = r.json()
        if j.get("status") != "OK":
            return {"ok": False, "error": j}
        return {"ok": True, "results": j["results"]}
    except Exception as e:
        return {"ok": False, "error": str(e)}

def get_ip_info(ip: Optional[str] = None):
    try:
        url = "http://ip-api.com/json/"
        if ip:
            url += requests.utils.quote(ip)
        r = requests.get(url, timeout=8)
        r.raise_for_status()
        return {"ok": True, "raw": r.json()}
    except Exception as e:
        return {"ok": False, "error": str(e)}

def get_public_holidays(country_code: str, year: int):
    try:
        url = f"https://date.nager.at/api/v3/PublicHolidays/{year}/{country_code}"
        r = requests.get(url, timeout=10)
        r.raise_for_status()
        return {"ok": True, "holidays": r.json()}
    except Exception as e:
        return {"ok": False, "error": str(e)}

FUNCTION_MAP = {
    "query_sqlite": query_sqlite,
    "get_country_info": get_country_info,
    "get_weather": get_weather,
    "convert_currency": convert_currency,
    "get_exchange_rate": get_exchange_rate,
    "get_cat_fact": get_cat_fact,
    "get_random_joke": get_random_joke,
    "get_sun_times": get_sun_times,
    "get_ip_info": get_ip_info,
    "get_public_holidays": get_public_holidays,
}

Definir JSON Schemas de tools

Una vez que tienes las funciones Python implementadas, el modelo todavía no sabe qué herramientas existen ni cómo debe llamarlas, por lo que necesitas describirlas mediante JSON Schema. Esta descripción indica al modelo el nombre de la herramienta, en qué casos debería usarla y qué parámetros espera, con sus tipos y cuáles son obligatorios.

La lista TOOLS_JSON_SCHEMA actúa como un catálogo de herramientas que se pasa al SDK de OpenAI en el parámetro tools, y cada entrada debe tener un nombre que coincida con la clave correspondiente en FUNCTION_MAP. Gracias a esta descripción formal el modelo puede construir llamadas bien formadas, con argumentos válidos y alineados con lo que tu código espera, reduciendo errores y ambigüedades en la invocación de funciones.

TOOLS_JSON_SCHEMA = [
    {
        "type": "function",
        "name": "query_sqlite",
        "description": "Ejecuta una consulta SELECT a la base de datos SQLite de clientes,no se permite poner punto y coma, solo se permite SELECT, no se permite INSERT, UPDATE, DELETE ni sentencias de escritura.",
        "parameters": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "Consulta SELECT simple sobre la tabla 'customer'. Columnas disponibles: id, name, email, country, currency, phone, balance."}
            },
            "required": ["sql"]
        }
    },
    {
        "type": "function",
        "name": "get_country_info",
        "description": "Obtiene información de un país como capital y lat/lon",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Nombre o código del país"}
            },
            "required": ["query"]
        }
    },
    {
        "type": "function",
        "name": "get_weather",
        "description": "Obtiene la temperatura y condiciones actuales para lat/lon",
        "parameters": {
            "type": "object",
            "properties": {
                "latitude": {"type": "number"},
                "longitude": {"type": "number"}
            },
            "required": ["latitude", "longitude"]
        }
    },
    {
        "type": "function",
        "name": "convert_currency",
        "description": "Convierte un monto de una moneda a otra",
        "parameters": {
            "type": "object",
            "properties": {
                "amount": {"type": "number"},
                "from_currency": {"type": "string"},
                "to_currency": {"type": "string"}
            },
            "required": ["amount", "from_currency", "to_currency"]
        }
    },
    {
        "type": "function",
        "name": "get_exchange_rate",
        "description": "Obtiene tasas de cambio de una moneda base a varias monedas",
        "parameters": {
            "type": "object",
            "properties": {
                "base": {"type": "string"},
                "symbols": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["base"]
        }
    },
    {
        "type": "function",
        "name": "get_cat_fact",
        "description": "Devuelve un dato curioso sobre gatos",
        "parameters": {"type": "object", "properties": {}}
    },
    {
        "type": "function",
        "name": "get_random_joke",
        "description": "Devuelve un chiste aleatorio",
        "parameters": {"type": "object", "properties": {}}
    },
    {
        "type": "function",
        "name": "get_sun_times",
        "description": "Devuelve hora de salida y puesta de sol para lat/lon",
        "parameters": {
            "type": "object",
            "properties": {
                "latitude": {"type": "number"},
                "longitude": {"type": "number"},
                "date": {"type": "string"}
            },
            "required": ["latitude", "longitude"]
        }
    },
    {
        "type": "function",
        "name": "get_ip_info",
        "description": "Devuelve información geográfica de una IP",
        "parameters": {
            "type": "object",
            "properties": {
                "ip": {"type": "string"}
            }
        }
    },
    {
        "type": "function",
        "name": "get_public_holidays",
        "description": "Devuelve días festivos de un país en un año",
        "parameters": {
            "type": "object",
            "properties": {
                "country_code": {"type": "string"},
                "year": {"type": "integer"}
            },
            "required": ["country_code", "year"]
        }
    },
]

Probar manualmente las tools

Antes de dejar que el modelo utilice las herramientas en un bucle automático, es recomendable probar cada función manualmente desde Python para asegurarte de que hace exactamente lo que esperas. En este bloque inicializas la base de datos de clientes, lanzas una consulta general, filtras por una clienta concreta y utilizas sus datos para encadenar distintas tools.

El flujo parte de Emma Johnson, recuperando su registro con query_sqlite, después obtiene información de su país con get_country_info, usa las coordenadas para llamar a get_weather y get_sun_times, convierte su saldo a euros con convert_currency y finalmente consulta los festivos con get_public_holidays. Estas pruebas manuales te permiten validar que cada herramienta devuelve estructuras coherentes antes de introducir al LLM en la ecuación.

print("\n--- Inicializando DB ---")
init_db(force=True)

print("\n--- query_sqlite ---")
resp = query_sqlite("SELECT * FROM customer")
print(resp)

print("\n--- query_sqlite Emma ---")
resp = query_sqlite("SELECT * FROM customer WHERE email='emma.johnson@example.com'")
print(resp)
emma = resp["rows"][0] if resp.get("rows") else None

print("\n--- get_country_info ---")
if emma:
    country_info = get_country_info(emma["country"])
    print(country_info)

    print("\n--- get_weather ---")
    latlng = country_info.get("latlng")
    if latlng:
        weather = get_weather(latlng[0], latlng[1])
        print(weather)

    print("\n--- convert_currency ---")
    converted = convert_currency(emma["balance"], emma["currency"], "EUR")
    print(converted)

print("\n--- get_exchange_rate ---")
rates = get_exchange_rate("USD", ["EUR","GBP"])
print(rates)

# print("\n--- get_cat_fact ---")
# print(get_cat_fact())

# print("\n--- get_random_joke ---")
# print(get_random_joke())

if latlng:
    print("\n--- get_sun_times ---")
    print(get_sun_times(latlng[0], latlng[1]))

# print("\n--- get_ip_info ---")
# print(get_ip_info())

print("\n--- get_public_holidays ---")
print(get_public_holidays("ES", 2025))

Implementación del bucle

Con las herramientas listas, el siguiente paso es construir el bucle de herramientas, que es la pieza que convierte al modelo en un agente capaz de decidir qué tool usar en cada momento. La función run_tool_loop recibe el prompt del usuario, inicializa una lista de mensajes input_list y controla el número máximo de llamadas a herramientas mediante el contador call_count.

En cada iteración llamas a client.responses.create pasando el modelo, la lista de tools y todo el historial acumulado en input_list, y de la respuesta extraes response.output, que puede contener tanto mensajes normales como llamadas a funciones. A partir de ahí filtras los elementos cuyo type es "function_call", ejecutas la función Python correspondiente buscándola en FUNCTION_MAP, y devuelves su resultado al modelo como elementos de tipo "function_call_output" con el mismo call_id, permitiendo así que el modelo los relacione correctamente y siga razonando paso a paso.

def run_tool_loop(user_prompt: str, max_calls: int = 20):
    # init_db(force=False)
    input_list = [{"role":"user","content":user_prompt}]
    call_count = 0

    while call_count < max_calls:
        response = client.responses.create(
            model="gpt-5.1",
            tools=TOOLS_JSON_SCHEMA,
            input=input_list
        )

        output_items = response.output
        input_list += output_items

        function_calls = [item for item in output_items if getattr(item,"type",None)=="function_call"]

        if not function_calls:
            break

        for item in function_calls:
            name = getattr(item,"name",None)
            call_id = getattr(item,"call_id",None)
            args_json = getattr(item,"arguments","{}")
            try:
                args = json.loads(args_json)
            except:
                args = {}

            func = FUNCTION_MAP.get(name)
            if not func:
                input_list.append({
                    "type":"function_call_output",
                    "call_id":call_id,
                    "output":json.dumps({"ok":False,"error":f"Function {name} no implementada"})
                })
                call_count +=1
                continue

            try:
                if isinstance(args, dict):
                    result = func(**args)
                else:
                    result = func(args)
            except Exception as e:
                result = {"ok": False, "error": str(e)}

            input_list.append({
                "type":"function_call_output",
                "call_id": call_id,
                "output": json.dumps(result)
            })
            call_count += 1

    return input_list

Ejecutar e imprimir la conversación

En el último paso construyes un prompt que describe quién es la usuaria y qué necesita, de forma que el modelo tenga que encadenar varias herramientas para resolver la petición de principio a fin. Después llamas a run_tool_loop, recoges toda la conversación devuelta y la recorres imprimiendo cada mensaje, lo que te permite inspeccionar cómo ha ido usando las tools el agente y depurar fácilmente su comportamiento.

prompt = "Soy Emma Johnson, mi email es emma.johnson@example.com, indica cuál es mi saldo en EUR, y qué temperatura hace en la capital de mi país y qué festivos hay en mi país para el año 2025."
conversation = run_tool_loop(prompt, max_calls=20)

for message in conversation:
    print(message)
    print()
Alan Sastre - Autor del tutorial

Alan Sastre

Ingeniero de Software y formador, CEO en CertiDevs

Ingeniero de software especializado en Full Stack y en Inteligencia Artificial. Como CEO de CertiDevs, OpenAI es una de sus áreas de expertise. Con más de 15 años programando, 6K seguidores en LinkedIn y experiencia como formador, Alan se dedica a crear contenido educativo de calidad para desarrolladores de todos los niveles.

Más tutoriales de OpenAI

Explora más contenido relacionado con OpenAI y continúa aprendiendo con nuestros tutoriales gratuitos.

Aprendizajes de esta lección

Cursos que incluyen esta lección

Esta lección forma parte de los siguientes cursos estructurados con rutas de aprendizaje