QVeris Python SDK v0.7.3 是最新测试版本。使用异步客户端,在你自己的 Agent 和应用中发现、检查、探测、调用并审计 丰富的 API 能力。
SDK 提供两种控制粒度:
QverisClient —— QVeris REST API 的轻量类型化封装(discover、inspect、probe、call、usage、ledger)。Agent —— 开箱即用的 LLM 工具循环,让模型自主发现并调用能力。需要完全控制时用 client;想几行代码就跑起来一个可用助手时用 agent。
pip install qveris
需要 Python 3.8+。运行时依赖:httpx、pydantic、pydantic-settings、openai。
SDK 从环境变量 QVERIS_API_KEY 读取 API 密钥,并通过 QVERIS_BASE_URL 指向 API 地址。请设置:
export QVERIS_API_KEY="your-api-key"
export QVERIS_BASE_URL="https://qveris.cn/api/v1"
在控制台/API密钥中创建密钥。也可以显式传入配置:
from qveris import QverisClient, QverisConfig
client = QverisClient(QverisConfig(
api_key="your-api-key",
base_url="https://qveris.cn/api/v1",
))
地址优先级为:显式 base_url > QVERIS_BASE_URL。API key 不参与地址选择;请按上面的示例设置地址。覆盖值必须是无凭据、查询串和片段的 HTTP(S) URL。
默认流程是 discover(发现)→ call(调用),之后可选 audit(审计)。inspect 和 probe 是按需检查,不是必经步骤。所有方法都是 async。
进行 Provider 比较时,如果需要确认当前范围或完整契约,必须逐一 Inspect;Discover 摘要不等于确认。比较需要当前报价时,必须逐一 Probe。复用只能保留精确路由,不能保留业务参数或结果:参数必须来自当前请求;当前、最新、今天或其他时效性数据必须执行新的 Call。
import asyncio
import math
from qveris import QverisClient, QverisConfig
def matches_type(kind, value):
return {
"string": lambda: isinstance(value, str),
"integer": lambda: isinstance(value, int) and not isinstance(value, bool),
"number": lambda: isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value),
"boolean": lambda: isinstance(value, bool),
"array": lambda: isinstance(value, list),
"object": lambda: isinstance(value, dict),
}.get(kind, lambda: False)()
def supports_request(candidate, params):
if candidate.params is None:
return False
definitions = {p.name: p for p in candidate.params}
if len(definitions) != len(candidate.params):
return False
return all(not p.required or p.name in params for p in candidate.params) and all(
(p := definitions.get(name)) is not None
and matches_type(p.type, value)
and (p.enum is None or any(allowed == value and
isinstance(allowed, bool) == isinstance(value, bool) for allowed in p.enum))
for name, value in params.items()
)
async def main():
client = QverisClient(QverisConfig(base_url="https://qveris.cn/api/v1"))
try:
# 1. 用自然语言发现能力(免费)
discovered = await client.discover("天气预报 API", limit=5)
params = {"city": "北京"}
tool = next(
(candidate for candidate in discovered.results
if supports_request(candidate, params)),
None,
)
# 2. 仅在 Discover 未提供选择所需契约时 Inspect
if tool is None:
details = await client.inspect(
[candidate.tool_id for candidate in discovered.results[:3]],
search_id=discovered.search_id,
)
tool = next(
(candidate for candidate in details.results
if supports_request(candidate, params)),
None,
)
if tool is None:
raise RuntimeError("没有候选能力提供兼容的当前参数契约")
# 3. 样例只作模板;覆盖为本次请求的真实业务值
result = await client.call(
tool.tool_id,
params,
search_id=discovered.search_id,
max_response_size=20480,
)
print(result.success, result.result)
# 4. 审计最终扣费结果
usage = await client.usage(execution_id=result.execution_id, summary=True)
ledger = await client.ledger(summary=True, limit=5)
print(usage.total, ledger.total)
finally:
await client.close()
asyncio.run(main())
QverisClient持有一个 HTTP 连接池。用完务必await client.close()(建议放在finally中)。
客户端在路由层面是无状态的,没有隐式语义路由、schema、费用或结果缓存。应在当前应用流程中保留真实 search_id。Host 如自行实现复用,必须按账户/API 地址/授权/会话隔离,根据当前请求重建业务值,并显式管理元数据失效。
显式空参数列表表示工具确实无参数;缺少参数列表表示当前投影没有提供契约。仅在选择或构造合法请求所需详情缺失/过期,或需要比较候选时使用 inspect。仅在参数需要校验、预算决策需要当前报价、或明确要求预检时使用 probe。报价不等于锁价,也不能代替用户授权。
若已设置
QVERIS_BASE_URL=https://qveris.cn/api/v1环境变量,可直接QverisClient(),无需再传base_url。
Agent 把同一套流程封装成一个 LLM 工具循环。模型会拿到 discover、inspect、call 三个工具;默认指引要求 Discover 后直接 Call,只在契约详情缺失/过期或需要比较候选时使用 Inspect。
每次内置 call 都会自动将 AgentConfig.model 记录为 Call 的 model 归因。该值属于 Agent 自身的元数据,因此模型生成的工具参数不能省略或覆盖它。
默认 agent 使用 OpenAI 兼容的 provider。配合国内可用的 OpenAI 兼容服务时,设置:
export OPENAI_API_KEY="..."
export OPENAI_BASE_URL="https://your-openai-compatible-endpoint/v1" # 你的 OpenAI 兼容服务地址
QVeris API 地址仍由 QVERIS_BASE_URL 控制,与 LLM provider 相互独立。
import asyncio
from qveris import Agent, Message, QverisConfig
async def main():
async with Agent(config=QverisConfig(base_url="https://qveris.cn/api/v1")) as agent:
messages = [Message(role="user", content="查一下北京现在的天气。")]
async for event in agent.run(messages):
if event.type == "content" and event.content:
print(event.content, end="", flush=True)
asyncio.run(main())
Agent 是异步上下文管理器 —— async with Agent() as agent: 会自动释放网络资源。
只需要最终回答时:
async with Agent(config=QverisConfig(base_url="https://qveris.cn/api/v1")) as agent:
answer = await agent.run_to_completion(
[Message(role="user", content="找一个股票报价能力并查询 AAPL。")]
)
print(answer)
Agent.run(messages) 产出 StreamEvent 对象。通过 event.type 区分:
type | 含义 |
|---|---|
content | 助手文本(流式时为增量分片,否则为完整消息) |
reasoning / reasoning_details | 部分 provider 的推理 token / 结构化推理 |
tool_call | 模型正在调用 discover / inspect / call(或你的额外工具) |
tool_result | 工具调用的执行结果(event.tool_result 含 name、result、is_error) |
metrics | provider 上报的 token 用量 / 耗时 |
error | 终止本次运行的致命错误 |
budget_warning / budget_exceeded | 会话消耗越过预警阈值 / 某次调用因超预算被拦截(见 预算护栏) |
给 run(...) 传 stream=False 可改为接收完整助手轮次而非增量分片。
设置会话级积分预算以约束自主消耗:
agent = Agent(budget_credits=25)
设置后,agent 会从 discover / inspect 学习每个能力的 expected_cost,并在请求发出之前拦截预计会超预算的 call——同时发出 budget_exceeded 事件,让模型改选更便宜的能力或停止。它从 call 的账单累计实际消耗,并在接近上限时发出一次 budget_warning。可随时查询状态:
status = agent.budget_status() # {"limit": 25, "spent": 12.0, "remaining": 13.0},或 None
spent 反映预结算金额——请用 usage(...) / ledger(...) 核对最终扣费。成本未知的能力不会被拦截(无法估算)。未设置 budget_credits 时,agent 行为完全不变。
QverisConfig| 字段 | 环境变量 | 默认值 | 说明 |
|---|---|---|---|
api_key | QVERIS_API_KEY | None | API 密钥,以 Authorization: Bearer ... 发送 |
base_url | QVERIS_BASE_URL | 按上文设置 | API 基础地址 |
credential_audience | QVERIS_CREDENTIAL_AUDIENCE | None | 透传给凭据 provider 的可选 audience |
credential_scopes | QVERIS_CREDENTIAL_SCOPES | () | 透传给凭据 provider 的可选 scopes |
max_retries | QVERIS_MAX_RETRIES | 3 | 读/审计操作的有界 429/503 重试;不适用于付费调用 |
read_timeout | QVERIS_READ_TIMEOUT | 30 | 读/审计操作的默认 HTTP 超时(秒) |
call_timeout | QVERIS_CALL_TIMEOUT | 120 | 付费调用的默认 HTTP 超时(秒) |
enable_history_pruning | — | True | 裁剪/压缩旧的工具输出以节省 token(agent 循环) |
max_iterations | — | 50 | agent 工具循环的最大迭代次数 |
已登记的机密 Agent Runtime 可使用 AgentDelegationCredentialProvider,通过
https://qveris.cn/api/v1/oauth/token 换取委托 token。将
credential_audience 设为委托 resource,并把 credential_scopes 设为客户端所需
scope。提供器只在内存缓存短期 token、不刷新 token,并拒绝 audience、scope、预算、
工具、provider、run 或 model 扩大。机密客户端 secret 必须留在可信服务端。
AgentConfig| 字段 | 默认值 | 说明 |
|---|---|---|
model | gpt-4o | 传给当前 LLM provider 的模型名 |
additional_system_prompt | None | 追加到默认工具使用系统提示词之后 |
temperature | 0.7 | 在 provider 支持时透传 |
from qveris import Agent, QverisConfig, AgentConfig
agent = Agent(
config=QverisConfig(base_url="https://qveris.cn/api/v1", max_iterations=20),
agent_config=AgentConfig(model="gpt-4o", temperature=0.2),
)
付费 call() 默认严格 single-submit。SDK 不会跟随 HTTP 重定向,不会自动重试 429/503、超时或 transport 失败,也不会删除被拒绝的投影字段后再次提交;类型化错误会报告 request_metadata.http_attempts == 1。如果旧服务仍需要原来的投影降级,可显式选择:
result = await client.call(
"tool.id",
{"symbol": "AAPL"},
respond_with="summary",
compatibility_mode="legacy_optional_fields", # 已弃用;可能提交两次
)
该模式会发出 DeprecationWarning,并用 request_metadata.compatibility_replays 记录重放。读和审计操作仍使用有界 max_retries。
QverisClient 可以注入共享 http_client,或配置由 SDK 持有的 transport / limits;两种形式互斥。close() 会等待在途操作、支持并发幂等调用、只关闭 SDK 自己创建的 client;关闭后新请求会抛出 QverisClientClosedError。
每次物理 attempt 都会给凭据 provider 传入不可变 CredentialContext,其中包含 resource、显式配置的 audience/scopes、operation、purpose、session ID 和可选的非敏感 correlation_id。HTTP 超时从凭据获取完成后开始计算。
公开失败统一使用 QverisError 层级(QverisApiError、QverisTransportError、QverisCredentialError、QverisContractError、QverisClientClosedError)。错误会保留安全的机器字段和不可变 RequestMetadata,但不会保留原始 HTTP request/response、bearer credential、签名 URL 或底层异常对象。
max_retries 仅用于读和审计操作,不再控制付费调用。call(..., respond_with=...) 不再静默执行旧版重放;只应在短期迁移窗口使用 compatibility_mode="legacy_optional_fields"。QverisError 或其类型化子类,不再捕获 httpx.HTTPStatusError / 原始 transport 异常。response.request_metadata,不会进入 wire 序列化。Capability Resolve/Query、selection token、idempotency key 和 execution lookup 只会在对应端点及字段进入公开 OpenAPI 后提供;client 不会提前发明临时 wire 字段。
以下章节概述当前公开 client、Agent、配置和响应模型的签名。
QverisClient| 方法 | REST 端点 | 用途 |
|---|---|---|
discover(query, limit=20, session_id=None, view=None, lang=None, timeout=None, correlation_id=None) | POST /search | 发现能力;view="routing" 返回精简 routing card(免费) |
inspect(tool_ids, search_id=None, session_id=None, timeout=None, correlation_id=None) | POST /tools/by-ids | 获取能力完整元数据(免费) |
probe(tool_id, parameters=None, checks=None, live_budget="none", timeout=None, correlation_id=None) | POST /tools/probe | 校验参数并请求零成本报价 |
call(tool_id, parameters, ..., model=None, compatibility_mode="strict", timeout=None, correlation_id=None) | POST /tools/execute | 使用严格 single-submit 语义执行能力,并可记录模型归因 |
usage(**filters) | GET /auth/usage/history/v2 | 审计请求状态与扣费结果 |
ledger(**filters) | GET /auth/credits/ledger | 查看最终积分余额变动 |
handle_tool_call(func_name, func_args, session_id=None) | — | 把 LLM 工具调用桥接到对应的 QVeris 方法 |
close() | — | 关闭底层 HTTP 客户端 |
tool_ids 接受单个字符串或可迭代对象。usage(...) 和 ledger(...) 接受仅关键字过滤参数,如 start_date、end_date、summary(默认 True)、bucket、charge_outcome、execution_id、search_id、direction、entry_type、min_credits、max_credits、limit、page、page_size。
仍保留向后兼容别名:search_tools → discover,get_tools_by_ids → inspect,execute_tool → call。
投影参数仅在显式指定时发送。Discover 的读侧投影兼容仍然有界;付费调用只有在显式选择已弃用兼容模式时才会重放。无效投影仍按错误返回。
Agent| 成员 | 说明 |
|---|---|
run(messages, stream=True) | 产出 StreamEvent 的异步生成器;主集成 API |
run_to_completion(messages) | 非流式;返回最终助手文本 |
get_last_messages() | 上一次 run(...) 的对话历史,含工具调用/结果 |
new_session() | 重置关联/会话 id |
close() | 释放网络资源(或用 async with) |
构造函数:Agent(config=None, agent_config=None, llm_provider=None, extra_tools=None, extra_tool_handler=None, debug_callback=None)。
SDK 返回 Pydantic v2 模型,因此你能获得自动补全与校验。未知的后端字段会被保留,因此较新的 API 元数据不会破坏较旧的 SDK 客户端。
SearchResponse → results: list[ToolInfo];ToolInfo 含 tool_id、name、description、params: list[ToolParameter]、examples、stats、billing_rule。ToolExecutionResponse,含 execution_id、success、result、error_message、billing(CompactBillingStatement)、cost、remaining_credits。UsageHistoryResponse → items: list[UsageEventItem]、total、summary。CreditsLedgerResponse → items: list[CreditsLedgerItem]、total、summary。from qveris import ToolExecutionResponse
def explain(result: ToolExecutionResponse) -> str:
if not result.success:
return f"失败:{result.error_message}"
charged = result.billing.summary if result.billing else "无计费信息"
return f"成功({charged});剩余={result.remaining_credits}"
按你的应用选择合适的粒度:
discover/inspect/call/usage/ledger。Agent.run(messages) 并消费 StreamEvent。Agent.run(messages, stream=False),获取完整轮次与事件。Agent.run_to_completion(messages)。from qveris import QverisClient, QverisConfig
from qveris.client.tools import DISCOVER_TOOL_DEF, INSPECT_TOOL_DEF, CALL_TOOL_DEF
tools = [DISCOVER_TOOL_DEF, INSPECT_TOOL_DEF, CALL_TOOL_DEF]
client = QverisClient(QverisConfig(base_url="https://qveris.cn/api/v1"))
# ... 你的 LLM 产出一个工具调用 (func_name, func_args) ...
result, is_error, handled = await client.handle_tool_call(func_name, func_args)
if handled and not is_error:
... # 把 result 回传给模型
把 QVeris 的 discover/inspect/call 工作流暴露为主流 Agent 框架的原生工具。适配器惰性导入各自框架,因此基础 qveris 包不依赖它们。
| 框架 | 原生工具类型 | Adapter 安装 | 完整 Agent 还需要 |
|---|---|---|---|
| LangChain / LangGraph | StructuredTool | pip install "qveris[langchain]"(Adapter 支持 Python 3.9+) | 下方当前版 create_agent 示例需要 Python 3.10+、langchain>=1.0 和模型 provider 包。 |
| OpenAI Agents SDK | FunctionTool | pip install "qveris[openai-agents]"(Python 3.10+) | 将工具传给 Agent,最后 await client.close()。 |
| CrewAI | BaseTool | pip install "qveris[crewai]"(Python 3.10+) | Adapter 负责同步/异步桥接,最后调用 aclose(client)。 |
| AutoGen | autogen_core.tools.FunctionTool | pip install "qveris[autogen]"(Python 3.10+) | 另装 autogen-agentchat 和模型扩展,如 autogen-ext[openai]。 |
| LlamaIndex | llama_index.core.tools.FunctionTool | pip install "qveris[llamaindex]"(Python 3.10+) | 另装 FunctionAgent 使用的模型集成包;使用异步 Agent 或 await tool.acall(...)。 |
| Pydantic AI | pydantic_ai.Tool | pip install "qveris[pydantic-ai]"(Python 3.10+) | Extra 为精简安装;另装模型 provider extra,如 pydantic-ai-slim[openai]。 |
每次调用 get_qveris_tools(client, session_id=...) 都会按顺序返回 qveris_discover、qveris_inspect、qveris_call 三个工具。工具结果(包括 QVeris 错误结果)统一为 JSON 字符串,Agent 可以读取并自行调整参数或改选能力。discover 免费并返回 search_id,后续应把它传给 inspect 和 call。完整 Agent 运行既需要 QVERIS_API_KEY,也需要所选模型 provider 的 API key。
LangChain 与 LangGraph
使用 LangChain 当前的 create_agent API。它运行在 LangGraph 上;自定义 LangGraph 工作流也可以把同一组工具放入 ToolNode。
pip install "qveris[langchain]" "langchain>=1.0" langchain-openai
import asyncio
from langchain.agents import create_agent
from qveris import QverisClient, QverisConfig
from qveris.integrations.langchain import get_qveris_tools
async def main():
client = QverisClient(QverisConfig(base_url="https://qveris.cn/api/v1"))
try:
agent = create_agent("openai:gpt-4o-mini", tools=get_qveris_tools(client))
result = await agent.ainvoke({"messages": [{"role": "user", "content": "查找股票报价工具并查询 AAPL。"}]})
print(result)
finally:
await client.close()
asyncio.run(main())
OpenAI Agents SDK
import asyncio
from agents import Agent, Runner
from qveris import QverisClient, QverisConfig
from qveris.integrations.openai_agents import get_qveris_tools
async def main():
client = QverisClient(QverisConfig(base_url="https://qveris.cn/api/v1"))
try:
agent = Agent(name="Assistant", tools=get_qveris_tools(client))
result = await Runner.run(agent, "查找股票报价能力并查询 AAPL。")
print(result.final_output)
finally:
await client.close()
asyncio.run(main())
CrewAI
from crewai import Agent
from qveris import QverisClient, QverisConfig
from qveris.integrations.crewai import aclose, get_qveris_tools
client = QverisClient(QverisConfig(base_url="https://qveris.cn/api/v1"))
agent = Agent(role="Researcher", goal="选择正确能力", backstory="工具专家", tools=get_qveris_tools(client))
# Crew(...).kickoff()
aclose(client)
CrewAI 的 client 连接运行在 Adapter 的专用事件循环中,因此要使用 aclose(client),不能使用 await client.close()。
AutoGen、LlamaIndex 与 Pydantic AI
# 以下片段假设已配置 QverisClient 以及框架所需的 model_client / llm;
# 只需选择与你的 Agent 框架对应的 Adapter。
# AutoGen AssistantAgent
from autogen_agentchat.agents import AssistantAgent
from qveris.integrations.autogen import get_qveris_tools
agent = AssistantAgent("assistant", model_client=model_client, tools=get_qveris_tools(client))
# LlamaIndex FunctionAgent
from llama_index.core.agent.workflow import FunctionAgent
from qveris.integrations.llamaindex import get_qveris_tools
agent = FunctionAgent(tools=get_qveris_tools(client), llm=llm)
# Pydantic AI Agent
from pydantic_ai import Agent
from qveris.integrations.pydantic_ai import get_qveris_tools
agent = Agent("openai:gpt-4o-mini", tools=get_qveris_tools(client))
完整的 provider 配置与资源关闭方式见下方可运行示例。TypeScript SDK 提供 Vercel AI SDK 适配器。
默认 Agent() 使用内置的 OpenAI 兼容 provider。如需对接其他模型 API,实现 LLMProvider 并传入:
from typing import AsyncGenerator, List
from openai.types.chat import ChatCompletionToolParam
from qveris import Agent
from qveris.config import AgentConfig
from qveris.llm.base import LLMProvider
from qveris.types import ChatResponse, Message, StreamEvent
class MyProvider(LLMProvider):
async def chat_stream(self, messages: List[Message], tools: List[ChatCompletionToolParam], config: AgentConfig) -> AsyncGenerator[StreamEvent, None]:
...
async def chat(self, messages: List[Message], tools: List[ChatCompletionToolParam], config: AgentConfig) -> ChatResponse:
...
agent = Agent(llm_provider=MyProvider())
httpx.HTTPStatusError;业务失败信封抛出 RuntimeError。error 类型的 StreamEvent 暴露并结束本次运行。run_to_completion(...) 会将其重新抛为 RuntimeError。result.success 只反映能力调用本身。不要把它当作最终扣费结论 —— 用 usage(...) / ledger(...) 确认扣费。可运行示例位于 packages/python-sdk/examples/:
| 示例 | 场景 |
|---|---|
finance_research.py | 股票报价 / 市场数据研究 |
risk_compliance.py | 制裁、负面舆情、合规筛查 |
crypto_market.py | 加密货币价格与成交量数据 |
data_analysis.py | 用外部能力数据丰富数据集 |
explainable_routing.py | 基于 why_recommended / expected_cost 的成本感知能力选型 |
budget_guard.py | 用 Agent(budget_credits=...) 设置会话级积分预算 |
agent_loop_integration.py | LLM agent 循环集成 |
interactive_chat.py | 交互式流式终端聊天 |
stock_debate.py | 多 Agent 股票研究辩论 |
langchain_integration.py | 把 QVeris 能力作为 LangChain 工具(qveris[langchain]) |
openai_agents_integration.py | 把 QVeris 能力作为 OpenAI Agents SDK 工具(qveris[openai-agents]) |
crewai_integration.py | 把 QVeris 能力作为 CrewAI 工具(qveris[crewai]) |
autogen_integration.py | 把 QVeris 能力作为 AutoGen 工具(qveris[autogen]) |
llamaindex_integration.py | 把 QVeris 能力作为 LlamaIndex 工具(qveris[llamaindex]) |
pydantic_ai_integration.py | 把 QVeris 能力作为 Pydantic AI 工具(qveris[pydantic-ai]) |
otel_tracing.py | 为 discover/call 生成 OpenTelemetry span(qveris[otel]) |
设置 QVERIS_API_KEY 后,能力示例会运行 discover/inspect;仅当设置 RUN_QVERIS_CALLS=1 时才执行 call。运行前请确保已设置 QVERIS_BASE_URL=https://qveris.cn/api/v1。
>=3.8。qverispackages/python-sdk这个页面对你有帮助吗?