dcp_ai.openai — OpenAI 集成¶
本文是英文版的参考译文,更新于 28/10/2026。如有出入,以英文版为准。标识符、命令和代码块保持原文,不作翻译。
用于 OpenAI Python 客户端的 DCP 包装器。它为每次聊天调用创建一个意图,并在调用返回时记录一个审计条目,为函数调用提供 DCP 工具,并让审计轨迹保持哈希链连接。
包装器保存未签名的记录,并且不评估任何策略:每次调用都会运行,其审计条目记录 policy_decision: "approved"。它接受的 secret_key 不会被使用。要获得已签名的 bundle 以及围绕函数调用的策略关卡,请参照 docs/QUICKSTART_OPENAI.md。
支持的 DCP 规范¶
DCP-01 至 DCP-09,部分支持:采用这些规范形式的未签名记录和草稿,以及通过 dcp_verify_bundle 进行的 bundle 验证。
它并未完整实现这些规范。
安装¶
已发布到 PyPI(26/10/2026)。 dcp-ai 2.10.0 包含本文描述的实现。请从 2.9.0 升级并更新环境或锁定文件。见 Python 发布说明。
2.9.0 将所有 bundle 发送到 V1。2.10.0 验证 V2,并对两种版本强制执行验证策略,包括异步入口 averify_bundle_tool_call。工具描述和审计记录本身不是签名证据或授权。
源码开发:
git clone https://github.com/dcp-ai-protocol/dcp-ai.git
pip install -e "/path/to/dcp-ai/sdks/python[openai]"
快速开始¶
from openai import OpenAI
from dcp_ai.openai import DCPOpenAIClient
client = OpenAI() # reads OPENAI_API_KEY from the environment
# The wrapper reads agent_id from the passport and human_id from the RPR;
# it does not check or sign them.
passport = {"agent_id": "agent:gpt-assistant-001", "capabilities": ["browse", "api_call"], "risk_tier": "medium"}
rpr = {"human_id": "rpr:human-001"}
dcp_client = DCPOpenAIClient(openai_client=client, passport=passport, rpr=rpr)
# Without tools of your own, the DCP tools are added (auto_intent=True), so the
# model may answer with tool calls instead of text.
response = dcp_client.chat_completions_create(
model="gpt-4",
messages=[{"role": "user", "content": "Summarize today's news"}],
)
message = response.choices[0].message
print(message.content or message.tool_calls)
for entry in dcp_client.get_audit_trail():
print(f"[{entry['timestamp']}] {entry['intent_id']} -> {entry['outcome']}")
API 参考¶
DCPOpenAIClient¶
DCPOpenAIClient(
openai_client: Any, # OpenAI() instance
passport: dict[str, Any], # agent_id and, for commission, capabilities and risk_tier are read
rpr: dict[str, Any], # human_id is read
secret_key: str = "", # Accepted but not used: nothing is signed
session_nonce: str = "", # A fresh one when empty
auto_intent: bool = True, # Adds DCP_TOOLS to calls that pass no tools
)
方法¶
| 方法 | 说明 |
|---|---|
chat_completions_create(**kwargs) |
创建一个意图,调用 client.chat.completions.create(**kwargs),并在其返回时记录一个审计条目。抛出异常的调用会传播该异常,不留下审计条目。 |
commission(purpose, capabilities?) |
启用代理(DCP-05 §3.1);返回未签名的证书并记录一个审计条目。证书和 commissioned 状态都是本地的:不会证明任何内容,也不会发送到别处。 |
get_audit_trail() |
以哈希链连接的审计条目:返回的是内存中的列表本身而非副本,因此修改它就会修改轨迹。 |
get_session_nonce() |
会话 nonce。 |
chat_completions_create 流程¶
- 创建一个意图:
action_type: "api_call"、通道api、estimated_impact: "low"、data_classes: ["none"]。 - 当
auto_intent开启且调用未传入tools时,添加DCP_TOOLS。 - 执行
client.chat.completions.create()。不做任何策略决定:调用总会运行。 - 如果调用返回,记录一个审计条目,带有
policy_decision: "approved",并将响应的id作为evidence.result_ref,链接到前一条目。
DCP_TOOLS¶
用于 DCP 的 OpenAI 函数调用工具定义:
# Fragment: client as in the Quickstart
from dcp_ai.openai import DCP_TOOLS
messages = [{"role": "user", "content": "Declare your intent to make an API call"}]
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=DCP_TOOLS,
)
可用工具¶
| 工具 | 参数 | 说明 |
|---|---|---|
dcp_verify_bundle |
signed_bundle_json: string |
验证 DCP Signed Bundle |
dcp_declare_intent |
action_type: enum, target_channel: enum, data_classes?: string[], estimated_impact?: enum |
在操作之前声明意图(DCP-02) |
dcp_commission_agent |
purpose: string, capabilities?: string[], risk_tier?: enum |
启用代理(DCP-05 §3.1) |
dcp_report_vitality |
task_completion_rate, error_rate, human_satisfaction, policy_alignment |
报告活力指标(DCP-05 §4.1) |
dcp_declare_rights |
rights: string[], jurisdiction: string |
声明代理的权利(DCP-08 §3.1) |
dcp_create_mandate |
human_id: string, authority_scope: string[], valid_from?: string, valid_until?: string |
创建委托授权(DCP-09 §3.1) |
action_type: browse, api_call, send_email, create_calendar_event, initiate_payment, update_crm, write_file, execute_code
target_channel: web, api, email, calendar, payments, crm, filesystem, runtime
estimated_impact: low, medium, high
handle_dcp_tool_call(tool_name, arguments, session_nonce="", verifier_policy=None)¶
处理 OpenAI 返回的 DCP 工具调用,并返回 JSON 字符串。除 dcp_verify_bundle 外,这些工具返回未签名的草稿(例如 dcp_declare_intent 返回 {"intent_declared": true, "intent": {...}},其中 agent_id 和 human_id 被设为 "self")。这些草稿不是可直接签名的规范性工件:"self" 短于
schema 对 agent_id 和 human_id 要求的 6 个字符,证书、权利和授权的草稿也缺少其 schema 的字段。
请调整它们,按 schema 校验,并用核心 SDK 签名。
dcp_verify_bundle 在 verifier_policy 下用 dcp_ai.v2 的 verify_signed_bundle_v2 验证 V2 bundle。策略可以是在
DEFAULT_VERIFIER_POLICY 之上设置 VerifierPolicy 字段的 dict(不带 risk_overrides 的 default_mode 适用于所有风险
等级),也可以是完整的 VerifierPolicy;默认为 DEFAULT_VERIFIER_POLICY。V1 bundle 只有 Ed25519 签名,没有
intended_verifier,也没有会话时间窗口,只有当策略具有 allow_v1_bundles、接受 ed25519(在 accepted_classical_algs 中,
advisory_rejected_algs 中没有该项)、不要求受众绑定、会话到期或会话最长时长,并且对其风险等级和安全等级会接受纯经典
bundle 时(spec/VERIFICATION.md V2-10)才会被接受;其工件必须是对象,护照的 agent_id 和 RPR 的 human_id 必须是字符串。
结果是 JSON,包含 verified、dcp_version、验证器的 errors 和 warnings(各最多 20 条,每条 500 个字符),对 V2 还包含
其 details(每个值以同样方式限制),bundle 提供的文本中的控制字符和格式字符会被转义。无论是否有正在运行的事件循环它都能
工作,并保持线程当前的事件循环不变;异步代码可以改为 await averify_bundle_tool_call(arguments, verifier_policy)。
dcp_verify_bundle 是在你的代码所传入策略下的密码学验证:verifier_policy 是你调用时的
参数,绝不会从模型的 arguments 中读取。它不是对已锚定委托人的准入,也不授权
任何操作(见 docs/ADMISSION.md)。
# Fragment: response is a chat completion with tool calls
import json
from dcp_ai.openai import handle_dcp_tool_call
for tool_call in response.choices[0].message.tool_calls or []:
if tool_call.type == "function" and tool_call.function.name.startswith("dcp_"):
result = handle_dcp_tool_call(
tool_call.function.name,
json.loads(tool_call.function.arguments),
)
print(result) # JSON with the result
高级示例 — 使用 DCP 的函数调用¶
# Fragment: passport and rpr as in the Quickstart
import json
from openai import OpenAI
from dcp_ai.openai import DCPOpenAIClient, handle_dcp_tool_call
dcp_client = DCPOpenAIClient(openai_client=OpenAI(), passport=passport, rpr=rpr)
# No tools passed: the DCP tools are added, so the model can call them
response = dcp_client.chat_completions_create(
model="gpt-4",
messages=[{"role": "user", "content": "Declare your intent to make an API call"}],
)
for tc in response.choices[0].message.tool_calls or []:
if tc.type == "function":
result = handle_dcp_tool_call(tc.function.name, json.loads(tc.function.arguments))
print(f"Tool: {tc.function.name} -> {result}")
开发¶
从检出的代码:
依赖项¶
dcp-ai— DCP SDK(crypto、merkle、verify、models)openai— 官方 OpenAI 客户端
许可证¶
Apache-2.0