pip install fetch-hive-sdk

export FETCH_HIVE_API_KEY=fhk_...

from fetch_hive_sdk import FetchHive

client = FetchHive()

client = FetchHive(api_key="fhk_...")

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"])

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

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"])

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())

client = FetchHive(
    api_key="fhk_...",
    base_url="https://api.fetchhive.com/v1",
    timeout=60,
)

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)