跳转到主要内容

​
使用 Python SDK 运行

当你想从 Python 中向智能体发送消息时,请使用官方的 fetch-hive-sdk 包。该 SDK 封装了公开的 POST /v1/agent/invoke 端点,处理身份验证,支持流式响应和多模态输入,并同时提供同步和 asyncio 两种变体。

​
安装

pip install fetch-hive-sdk
该 SDK 需要 Python 3.9+,并在底层使用 httpx。

​
身份验证

将 FETCH_HIVE_API_KEY 环境变量设置为你的工作区 API 密钥(SDK 会自动读取): 
export FETCH_HIVE_API_KEY=fhk_...

from fetch_hive_sdk import FetchHive

client = FetchHive()
或显式传入密钥: 
client = FetchHive(api_key="fhk_...")
请参阅 API Keys 了解如何创建和轮换密钥。

​
基本示例

向智能体发送消息并读取最终响应: 
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"])
请参阅非流式响应结构。

​
方法参考

参数类型必填说明
agentstr是智能体 ID
messagestr是你想发送的消息
thread_idstr否标识持久会话线程的任意字符串。Fetch Hive 会在首次使用时创建该线程,并在后续调用中恢复该线程。
messageslist[dict]否调用方管理的对话历史。每一项格式为:{"role": "user" | "assistant" | "system", "content": str}。
image_urlslist[str]否用于多模态输入、附加到当前 message 的 HTTPS 图像 URL
userstr否在用户追踪中显示的不透明调用方标识符
metadatadict[str, str | int | float | bool | None]否由调用方定义的扁平元数据,用于审计和日志筛选。请参阅调用元数据
该 SDK 为 invoke_agent 注入 streaming: false。要进行流式调用,请使用 invoke_agent_stream(见下文)。

​
处理响应

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

​
流式

使用 invoke_agent_stream 在事件到达时接收 Server-Sent Events。该方法返回一个生成器,会产出解析后的事件字典: 
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"])
该流会产出在 Invoke Agent → Response 中记录的相同事件类型:summary(在自动摘要触发时)、reasoning、response、tool、最终的 usage 事件,或在提供商在流过程中失败时的 error 事件。

​
异步流式

对于 asyncio 应用,使用 ainvoke_agent_stream。它的参数相同,但返回一个异步迭代器: 
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())

​
多轮对话

​
持久化线程

将任意字符串作为 thread_id 传入,Fetch Hive 会在首次调用时创建该线程,并在每次使用相同值的后续调用中恢复该线程: 
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",
)

​
无状态历史

自行管理状态,在 messages 中传入先前的回合。Fetch Hive 会将提供的历史用作上下文,但不会持久化: 
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."},
    ],
)

​
多模态输入

使用 image_urls 在当前消息上附加图像: 
result = client.invoke_agent(
    agent="vision-agent",
    message="Describe this image",
    image_urls=["https://example.com/photo.jpg"],
)
print(result["response"])
URL 必须以 https:// 开头。

​
配置

选项默认值说明
api_keyFETCH_HIVE_API_KEY 环境变量控制台中的 Bearer 令牌
base_urlhttps://api.fetchhive.com/v1覆盖 API 基础 URL
timeout120请求超时(秒)
client = FetchHive(
    api_key="fhk_...",
    base_url="https://api.fetchhive.com/v1",
    timeout=120,
)

​
错误

非 2xx 响应会引发 httpx.HTTPStatusError,其中包含状态码和响应体: 
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)
请参阅错误和速率限制了解状态码含义。

​
链接

​
下一步