Skip to main content
Use the official fetch-hive-sdk package when you want to invoke a prompt deployment from Python. The SDK wraps the public POST /v1/prompt/invoke endpoint with idiomatic helpers, handles authentication, and parses streaming responses for you.

​
Installation

pip install fetch-hive-sdk
The SDK requires Python 3.9+ and uses httpx under the hood.

​
Authentication

Set the FETCH_HIVE_API_KEY environment variable to your workspace API key (the SDK reads it automatically): 
export FETCH_HIVE_API_KEY=fhk_...

from fetch_hive_sdk import FetchHive

client = FetchHive()
Or pass the key explicitly: 
client = FetchHive(api_key="fhk_...")
See API Keys for how to create and rotate keys.

​
Basic example

Invoke a prompt deployment and read the final response: 
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 is synchronous and returns the parsed JSON body once the prompt has completed. See the non-streaming response shape.

​
Method reference

ArgumentTypeRequiredDescription
deploymentstrYesThe prompt deployment name
variantstrNoThe deployment variant name
inputsdict[str, Any]NoKey-value pairs for the prompt variables
userstrNoOpaque caller identifier surfaced in User Tracking
metadatadict[str, str | int | float | bool | None]NoFlat caller-defined metadata for audit and log filtering. See Invoke metadata
The SDK injects streaming: false for invoke_prompt. To stream, use invoke_prompt_stream (below).

​
Handling the response

The non-streaming response is a plain dict: 
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

Use invoke_prompt_stream to receive Server-Sent Events as they arrive. The method returns a generator that yields parsed event dicts: 
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"])
The stream yields the same event types documented in Invoke Prompt → Response: reasoning, response, a final usage event, or an error event if the provider fails mid-stream.

​
Async streaming

For asyncio applications, use ainvoke_prompt_stream. It has the same arguments but returns an async iterator: 
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())

​
Configuration

OptionDefaultDescription
api_keyFETCH_HIVE_API_KEY env varBearer token from the dashboard
base_urlhttps://api.fetchhive.com/v1Override the API base URL
timeout120Request timeout in seconds
client = FetchHive(
    api_key="fhk_...",
    base_url="https://api.fetchhive.com/v1",
    timeout=60,
)

​
Errors

Non-2xx responses raise an httpx.HTTPStatusError with the status code and response body. Wrap calls in try/except if you need to handle failures: 
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)
See Errors and Rate Limits for status code meanings.

​
Next steps