Skip to main content
Use the official fetch-hive-sdk package when you want to invoke a workflow deployment from Python. The SDK wraps the public POST /v1/workflow/invoke endpoint, handles authentication, and supports both direct responses and callback delivery.

​
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

Run a workflow deployment directly. The call blocks until the workflow finishes: 
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"])
See the direct response shape.

​
Method reference

ArgumentTypeRequiredDescription
deploymentstrYesThe workflow deployment name
variantstrNoThe deployment variant name
inputsdict[str, Any]NoKey-value pairs for the variables defined on the Start step
async_modeboolNoWhen True, return immediately and deliver the result by signed callback
callback_urlstrNoRequired when async_mode=True - the HTTPS callback URL to call when the run finishes
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
invoke_workflow builds the request body for you. When you pass async_mode=True, the SDK sends: 
{
  "async": { "enabled": true, "callback_url": "https://example.com/webhook" }
}

​
Handling the response

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

​
Callback delivery

Pass async_mode=True to return immediately and have Fetch Hive call your callback URL when the run finishes: 
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"])
Store the webhook_secret so you can verify the signature on the incoming callback. See Callback Delivery and Webhook Triggers for the verification flow and signed payload shape.

​
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 - increase for long-running direct workflow requests
client = FetchHive(
    api_key="fhk_...",
    base_url="https://api.fetchhive.com/v1",
    timeout=600,
)

​
Errors

Non-2xx responses raise an httpx.HTTPStatusError with the status code and response body: 
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)
See Error Handling for workflow-specific failure cases and Errors and Rate Limits for HTTP status code meanings.

​
Next steps