Saltar al contenido principal

​
Ejecutar con el SDK de Python

Usa el paquete oficial fetch-hive-sdk cuando quieras enviar un mensaje a un agente desde Python. El SDK envuelve el endpoint público POST /v1/agent/invoke, maneja la autenticación, admite streaming y entradas multimodales, y expone variantes sincrónicas y asyncio.

​
Instalación

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

​
Autenticación

Establece la variable de entorno FETCH_HIVE_API_KEY con la clave de API de tu espacio de trabajo (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 Claves de API para saber cómo crear y rotar claves.

​
Ejemplo básico

Envía un mensaje a un agente y lee la respuesta final: 
from fetch_hive_sdk import FetchHive

client = FetchHive()

reply = client.invoke_agent(
    agent="AGENT_UUID",
    message="Summarize the latest AI infrastructure trends",
)

print(reply["response"])
Consulta el formato de respuesta sin streaming.

​
Referencia del método

ArgumentoTipoRequeridoDescripción
agentstrSíEl ID del agente
messagestrSíEl mensaje que quieres enviar
thread_idstrNoUna cadena arbitraria que identifica un hilo de conversación persistente. Fetch Hive crea el hilo en el primer uso y lo reanuda en llamadas posteriores.
messageslist[dict]NoHistorial de conversación administrado por el llamador. Cada elemento: {"role": "user" | "assistant" | "system", "content": str}.
image_urlslist[str]NoURLs HTTPS de imágenes adjuntas al message actual para entradas multimodales
userstrNoIdentificador opaco del llamador expuesto en Seguimiento de usuarios
metadatadict[str, str | int | float | bool | None]NoMetadatos planos definidos por el llamador para auditoría y filtrado de registros. Consulta Metadatos de invocación
El SDK inyecta streaming: false para invoke_agent. Para hacer streaming, usa invoke_agent_stream (abajo).

​
Manejo de la respuesta

reply = client.invoke_agent(agent="AGENT_UUID", message="Hello")

print(reply["response"])      # final text
print(reply["model"])         # model identifier
print(reply["usage"])         # token usage breakdown
print(reply["request_id"])    # use this to look up the run in Logs
print(reply.get("tool_calls"))  # tool invocations made during the run

​
Streaming

Usa invoke_agent_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_agent_stream(
    agent="AGENT_UUID",
    message="Summarize the latest AI infrastructure trends",
    thread_id="user-456-support-session",
):
    if chunk.get("type") == "response":
        print(chunk.get("response", ""), end="", flush=True)
    elif chunk.get("type") == "tool":
        print(f"\n[Calling tool: {chunk.get('tool')}]")
    elif chunk.get("type") == "usage":
        print("\n\nUsage:", chunk["usage"])
El stream genera los mismos tipos de eventos documentados en Invocar Agente → Respuesta: summary (cuando se dispara el resumen automático), reasoning, response, tool, un evento final usage, o un evento error si el proveedor falla a mitad del stream.

​
Streaming asíncrono

Para aplicaciones con asyncio, usa ainvoke_agent_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_agent_stream(
        agent="AGENT_UUID",
        message="Hello",
        thread_id="user-456-support-session",
    ):
        if chunk.get("type") == "response":
            print(chunk.get("response", ""), end="", flush=True)

asyncio.run(main())

​
Conversaciones de múltiples turnos

​
Hilos persistentes

Pasa cualquier cadena como thread_id y Fetch Hive creará el hilo en la primera llamada y lo reanudará en llamadas posteriores con el mismo valor: 
client.invoke_agent(
    agent="AGENT_UUID",
    message="What are the main AI infrastructure trends right now?",
    thread_id="user-456-support-session",
)

client.invoke_agent(
    agent="AGENT_UUID",
    message="Which of those trends have the most enterprise adoption?",
    thread_id="user-456-support-session",
)

​
Historial sin estado

Administra el estado tú mismo pasando los turnos previos en messages. Fetch Hive usa el historial proporcionado como contexto pero no lo persiste: 
client.invoke_agent(
    agent="AGENT_UUID",
    message="Which of those trends have the most enterprise adoption?",
    messages=[
        {"role": "user", "content": "What are the main AI infrastructure trends right now?"},
        {"role": "assistant", "content": "Teams are focusing on evals, tool routing, and observability."},
    ],
)

​
Entradas multimodales

Adjunta imágenes al mensaje actual con image_urls: 
result = client.invoke_agent(
    agent="vision-agent",
    message="Describe this image",
    image_urls=["https://example.com/photo.jpg"],
)
print(result["response"])
Las URLs deben comenzar con https://.

​
Configuración

OpciónValor predeterminadoDescripción
api_keyvariable de entorno FETCH_HIVE_API_KEYToken Bearer del panel
base_urlhttps://api.fetchhive.com/v1Sobrescribe la URL base de la API
timeout120Tiempo de espera de la solicitud en segundos
client = FetchHive(
    api_key="fhk_...",
    base_url="https://api.fetchhive.com/v1",
    timeout=120,
)

​
Errores

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

try:
    reply = client.invoke_agent(agent="AGENT_UUID", message="Hello")
except httpx.HTTPStatusError as exc:
    print("Fetch Hive returned", exc.response.status_code, exc.response.text)
Consulta Errores y límites de tasa para conocer el significado de los códigos de estado.

​
Enlaces

​
Próximos pasos