Saltar al contenido principal

​
Ejecutar con el SDK de Python

Usa el paquete oficial fetch-hive-sdk cuando quieras invocar un deployment de flujo de trabajo desde Python. El SDK envuelve el endpoint público POST /v1/workflow/invoke, maneja la autenticación y admite tanto respuestas directas como entrega por callback.

​
Instalación

pip install fetch-hive-sdk
El SDK requiere Python 3.9+ y usa httpx por debajo.

​
Autenticación

Configura la variable de entorno FETCH_HIVE_API_KEY con tu clave de API del workspace (el SDK la lee automáticamente): 
export FETCH_HIVE_API_KEY=fhk_...

from fetch_hive_sdk import FetchHive

client = FetchHive()
O pasa la clave explícitamente: 
client = FetchHive(api_key="fhk_...")
Consulta API Keys para saber cómo crear y rotar claves.

​
Ejemplo básico

Ejecuta un deployment de flujo de trabajo directamente. La llamada se bloquea hasta que el flujo de trabajo termina: 
from fetch_hive_sdk import FetchHive

client = FetchHive()

run = client.invoke_workflow(
    deployment="YOUR_DEPLOYMENT_NAME",
    variant="YOUR_VARIANT_NAME",
    inputs={"topic": "State of enterprise AI in 2026"},
)

print(run["run_status"])
print(run["output"])
Consulta la forma de respuesta directa.

​
Referencia del método

ArgumentoTipoRequeridoDescripción
deploymentstrSíEl nombre del deployment del flujo de trabajo
variantstrNoEl nombre de la variante del deployment
inputsdict[str, Any]NoPares clave-valor para las variables definidas en el paso Start
async_modeboolNoCuando es True, devuelve respuesta inmediatamente y entrega el resultado mediante callback firmado
callback_urlstrNoRequerido cuando async_mode=True - la URL HTTPS de callback a la que llamar cuando la ejecución termine
userstrNoIdentificador opaco de quien llama, expuesto en User Tracking
metadatadict[str, str | int | float | bool | None]NoMetadata plana definida por quien llama, para auditoría y filtrado de registros. Consulta Metadata de invoke
invoke_workflow construye el cuerpo de la solicitud por ti. Cuando pasas async_mode=True, el SDK envía: 
{
  "async": { "enabled": true, "callback_url": "https://example.com/webhook" }
}

​
Manejo de la respuesta

run = client.invoke_workflow(deployment="my-workflow", variant="default")

print(run["run_status"])    # "completed" | "failed" | "running" | "queued"
print(run["output"])        # final workflow output
print(run["request_id"])    # use this to look up the run in Logs

​
Entrega por callback

Pasa async_mode=True para devolver respuesta inmediatamente y hacer que Fetch Hive llame a tu URL de callback cuando la ejecución termine: 
run = client.invoke_workflow(
    deployment="YOUR_DEPLOYMENT_NAME",
    variant="YOUR_VARIANT_NAME",
    inputs={"topic": "State of enterprise AI in 2026"},
    async_mode=True,
    callback_url="https://example.com/callback",
)

print("Queued:", run["run_status"])
print("Webhook secret:", run["webhook_secret"])
Almacena el webhook_secret para que puedas verificar la firma en el callback entrante. Consulta Entrega por callback y disparadores de webhook para el flujo de verificación y la forma del payload firmado.

​
Configuración

OpciónPredeterminadoDescripción
api_keyvariable de entorno FETCH_HIVE_API_KEYToken bearer del dashboard
base_urlhttps://api.fetchhive.com/v1Sobrescribe la URL base de la API
timeout120Timeout de la solicitud en segundos - aumenta para solicitudes directas de flujos de trabajo de larga duración
client = FetchHive(
    api_key="fhk_...",
    base_url="https://api.fetchhive.com/v1",
    timeout=600,
)

​
Errores

Las respuestas distintas a 2xx lanzan un httpx.HTTPStatusError con el código de estado y el cuerpo de la respuesta: 
import httpx

try:
    run = client.invoke_workflow(deployment="my-workflow", variant="default")
except httpx.HTTPStatusError as exc:
    print("Fetch Hive returned", exc.response.status_code, exc.response.text)
Consulta Manejo de errores para los casos de fallo específicos del flujo de trabajo y Errors and Rate Limits para el significado de los códigos de estado HTTP.

​
Enlaces

​
Siguientes pasos