Saltar al contenido principal

Ejecutar con el SDK de Python

Usa el paquete oficial fetch-hive-sdk cuando quieras invocar un deployment de prompt desde Python. El SDK envuelve el endpoint público POST /v1/prompt/invoke con ayudantes idiomáticos, maneja la autenticación y analiza las respuestas en streaming por ti.

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

Invoca un deployment de prompt y lee la respuesta final:
from fetch_hive_sdk import FetchHive

client = FetchHive()

result = client.invoke_prompt(
    deployment="YOUR_DEPLOYMENT_NAME",
    variant="YOUR_VARIANT_NAME",
    inputs={"text": "Fetch Hive helps teams ship AI products faster."},
)

print(result["response"])
invoke_prompt es síncrono y devuelve el cuerpo JSON analizado una vez que el prompt se ha completado. Consulta la forma de respuesta sin streaming.

Referencia del método

ArgumentoTipoRequeridoDescripción
deploymentstrEl nombre del deployment de prompt
variantstrNoEl nombre de la variante del deployment
inputsdict[str, Any]NoPares clave-valor para las variables del prompt
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
El SDK inyecta streaming: false para invoke_prompt. Para transmitir, usa invoke_prompt_stream (a continuación).

Manejo de la respuesta

La respuesta sin streaming es un dict simple:
result = client.invoke_prompt(deployment="my-prompt", variant="default")

print(result["response"])       # final text
print(result["model"])          # model identifier
print(result["usage"])          # token usage breakdown
print(result["request_id"])     # use this to look up the run in Logs

Streaming

Usa invoke_prompt_stream para recibir Server-Sent Events a medida que llegan. El método devuelve un generador que produce diccionarios de eventos analizados:
for chunk in client.invoke_prompt_stream(
    deployment="YOUR_DEPLOYMENT_NAME",
    variant="YOUR_VARIANT_NAME",
    inputs={"text": "Fetch Hive helps teams ship AI products faster."},
):
    if chunk.get("type") == "response":
        print(chunk.get("response", ""), end="", flush=True)
    elif chunk.get("type") == "usage":
        print("\n\nUsage:", chunk["usage"])
La transmisión produce los mismos tipos de evento documentados en Invoke Prompt → Response: reasoning, response, un evento usage final o un evento error si el proveedor falla durante la transmisión.

Streaming asíncrono

Para aplicaciones asyncio, usa ainvoke_prompt_stream. Tiene los mismos argumentos pero devuelve un iterador asíncrono:
import asyncio
from fetch_hive_sdk import FetchHive

async def main():
    client = FetchHive()
    async for chunk in client.ainvoke_prompt_stream(
        deployment="YOUR_DEPLOYMENT_NAME",
        variant="YOUR_VARIANT_NAME",
        inputs={"text": "Hello"},
    ):
        if chunk.get("type") == "response":
            print(chunk.get("response", ""), end="", flush=True)

asyncio.run(main())

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
client = FetchHive(
    api_key="fhk_...",
    base_url="https://api.fetchhive.com/v1",
    timeout=60,
)

Errores

Las respuestas distintas a 2xx lanzan un httpx.HTTPStatusError con el código de estado y el cuerpo de la respuesta. Envuelve las llamadas en try/except si necesitas manejar fallos:
import httpx

try:
    result = client.invoke_prompt(deployment="my-prompt", variant="default")
except httpx.HTTPStatusError as exc:
    print("Fetch Hive returned", exc.response.status_code, exc.response.text)
Consulta Errors and Rate Limits para el significado de los códigos de estado.

Enlaces

Siguientes pasos