跳转到主要内容

​
使用 Python SDK 运行

当你想从 Python 调用提示词部署时，请使用官方的 fetch-hive-sdk 包。SDK 使用符合 Python 风格的辅助函数包装了公共的 POST /v1/prompt/invoke 端点，处理身份验证，并为你解析流式响应。

​
安装

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 密钥。

​
基本示例

调用提示词部署并读取最终响应： 
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 是同步的，并在提示词完成后返回已解析的 JSON 主体。请参阅 非流式响应结构。

​
方法参考

参数类型必填描述
deploymentstr是提示词部署名称
variantstr否部署变体名称
inputsdict[str, Any]否提示词变量的键值对
userstr否在 用户跟踪 中显示的不透明调用方标识符
metadatadict[str, str | int | float | bool | None]否调用方定义的扁平元数据，用于审计和日志过滤。请参阅 调用元数据
SDK 会为 invoke_prompt 注入 streaming: false。要使用流式，请使用下面的 invoke_prompt_stream。

​
处理响应

非流式响应是一个普通的 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

​
流式响应

使用 invoke_prompt_stream 在到达时接收 Server-Sent Events。该方法返回一个生成器，逐个产出已解析的事件字典： 
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"])
该流会产生 调用提示词 → 响应 中记录的相同事件类型：reasoning、response、最终的 usage 事件，或在提供商在流中失败时产生的 error 事件。

​
异步流式

对于 asyncio 应用，请使用 ainvoke_prompt_stream。它具有相同的参数，但返回异步迭代器： 
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())

​
配置

选项默认值描述
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=60,
)

​
错误

非 2xx 响应会引发 httpx.HTTPStatusError，其中包含状态码和响应体。如需处理失败，请将调用包裹在 try/except 中： 
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)
有关状态码含义，请参阅 错误与速率限制。

​
链接

​
后续步骤