Skip to main content
Use the official fetch-hive-sdk package when you want to send a message to an agent from Python. The SDK wraps the public POST /v1/agent/invoke endpoint, handles authentication, supports streaming and multimodal inputs, and exposes both synchronous and asyncio variants.

​
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

Send a message to an agent and read the final response: 
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"])
See the non-streaming response shape.

​
Method reference

ArgumentTypeRequiredDescription
agentstrYesThe agent ID
messagestrYesThe message you want to send
thread_idstrNoAn arbitrary string identifying a persistent conversation thread. Fetch Hive creates the thread on first use and resumes it on subsequent calls.
messageslist[dict]NoCaller-managed conversation history. Each item: {"role": "user" | "assistant" | "system", "content": str}.
image_urlslist[str]NoHTTPS image URLs attached to the current message for multimodal inputs
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_agent. To stream, use invoke_agent_stream (below).

​
Handling the response

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

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

​
Async streaming

For asyncio applications, use ainvoke_agent_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_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())

​
Multi-turn conversations

​
Persistent threads

Pass any string as thread_id and Fetch Hive will create the thread on the first call and resume it on subsequent calls with the same value: 
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",
)

​
Stateless history

Manage state yourself by passing the previous turns in messages. Fetch Hive uses the supplied history for context but does not persist it: 
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."},
    ],
)

​
Multimodal inputs

Attach images to the current message with image_urls: 
result = client.invoke_agent(
    agent="vision-agent",
    message="Describe this image",
    image_urls=["https://example.com/photo.jpg"],
)
print(result["response"])
URLs must start with https://.

​
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=120,
)

​
Errors

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

​
Next steps