fetch-hive-sdk

Official Python SDK for Fetch Hive — invoke AI prompts, workflows, and agents from your application.

Installation

pip install fetch-hive-sdk

Quick start

from fetch_hive_sdk import FetchHive

client = FetchHive(api_key="fhk_...")
# or: client = FetchHive()  # reads FETCH_HIVE_API_KEY env var

Get your API key from the Fetch Hive dashboard.

Invoke a prompt

result = client.invoke_prompt(
    deployment="my-prompt",
    inputs={"name": "Alice", "topic": "machine learning"},
    metadata={},
)
print(result["response"])

Invoke a prompt (streaming)

for chunk in client.invoke_prompt_stream(
    deployment="my-prompt",
    inputs={"name": "Alice"},
):
    if chunk.get("type") == "response":
        print(chunk.get("response", ""), end="", flush=True)
    elif chunk.get("type") == "usage":
        print("\nUsage:", chunk["usage"])

Invoke a workflow

run = client.invoke_workflow(
    deployment="my-workflow",
    inputs={"customer_id": "42"},
    metadata={},
)
print(run["status"], run.get("output"))

Invoke a workflow (async)

run = client.invoke_workflow(
    deployment="my-workflow",
    inputs={"customer_id": "42"},
    async_mode=True,
    callback_url="https://example.com/webhook",
)
print("Queued:", run["run_id"])

Invoke an agent

reply = client.invoke_agent(
    agent="my-agent",
    message="What is the weather in London?",
    metadata={},
)
print(reply["response"])

Metadata

Pass optional metadata on prompt, workflow, or agent invokes to attach flat audit fields for log display and filtering. Metadata values must be strings, numbers, booleans, or None.

Invoke an agent (streaming)

for chunk in client.invoke_agent_stream(
    agent="my-agent",
    message="What is the weather in London?",
    thread_id="session-abc123",  # optional — persist conversation history
):
    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("\nUsage:", chunk["usage"])

Tool approvals

tool_approval is optional on invoke_agent, invoke_agent_stream, ainvoke_agent_stream, and invoke_workflow. Omit it, or pass "policy", to follow saved rules. "auto_approve" applies to this call only, and only when the API key is allowed to auto-approve actions.

When a response has status "awaiting_approval", decide with the API key. The run resumes on the server. Do not send decided_tool_approvals.

client.decide_tool_action_approval("WORKSPACE_ID", "APPROVAL_ID", "allow_once")

pending = client.list_tool_action_approvals("WORKSPACE_ID", status="pending")
one = client.get_tool_action_approval("WORKSPACE_ID", "APPROVAL_ID")

reply = client.invoke_agent(
    agent="my-agent",
    message="Send the follow-up",
    tool_approval="auto_approve",
)

decision is "allow_once" or "deny". always_allow is only available in the dashboard.

Multimodal (image) inputs

result = client.invoke_agent(
    agent="vision-agent",
    message="Describe this image",
    image_urls=["https://example.com/photo.jpg"],
)
print(result["response"])

Async streaming

import asyncio

async def main():
    async for chunk in client.ainvoke_agent_stream(
        agent="my-agent",
        message="Hello",
        thread_id="session-abc123",
    ):
        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("\nUsage:", chunk["usage"])

asyncio.run(main())

Authentication

Pass the API key to the constructor or set the environment variable:

export FETCH_HIVE_API_KEY=fhk_...

client = FetchHive()  # picks up FETCH_HIVE_API_KEY automatically

Configuration

Links

• Fetch Hive dashboard
• API documentation
• GitHub

Version

0.4.1

License

MIT — see LICENSE.