openai-agents-python 沙箱工作区写入负载解析：WritePayload 与 coerce_write_payload 源码深度剖析

openai-agents-python 沙箱工作区写入负载解析WritePayload 与 coerce_write_payload 源码深度剖析【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonWorkspace Payloads是 openai-agents-python 沙箱子系统中负责规范化写入沙箱工作区文件的数据负载的底层模块。当通过SandboxSession.write()向沙箱工作区写入文件时调用方传入的必须是定位在负载起始位置的二进制文件类对象file-like object该模块负责把这一约定落地为统一的数据结构与适配逻辑。阅读本文后你将理解WritePayload数据模型、coerce_write_payload转换入口、二进制读取适配器与内容长度推断链的完整实现并能据此正确编写符合沙箱写入契约的自定义流对象。一、模块定位沙箱写入契约的统一入口docs/ref/sandbox/session/workspace_payloads.md是 mkdocs 自动生成的 API 参考页其指向的源码模块位于 src/agents/sandbox/session/workspace_payloads.py。该模块很小但职责清晰为沙箱工作区写入提供一个统一的负载payload规范化层让所有后端Docker、Unix 本地沙箱等在write()时都能以相同的方式消费调用方传入的数据流。从源码结构看该模块与抽象基类 BaseSandboxSession.write() 的接口约定直接对应——抽象方法签名要求async def write( self, path: Path, data: io.IOBase, *, user: str | User | None None, ) - None: Write a file into the sessions workspace. :param path: Absolute path in the container or path relative to the workspace root. :param data: A file-like object positioned at the start of the payload. :param user: Optional sandbox user to perform the write as. 也就是说任何沙箱后端的write()都接收一个位于起始位置的二进制文件类对象而workspace_payloads模块负责把这个对象转化为可被传输层安全消费的WritePayload。二、核心数据结构WritePayload模块顶部的核心数据模型是一个冻结frozen数据类 WritePayloaddataclass(frozenTrue) class WritePayload: stream: io.IOBase content_length: int | None None两个字段的语义如下字段类型含义streamio.IOBase经过适配后的二进制可读流可安全调用read()/readinto()/seek()/tell()content_lengthint \| None负载的字节长度尽力推断可能为None供传输层预分配缓冲区或设置长度头frozenTrue意味着WritePayload创建后不可修改保证它在整个写入流程中传递时不会被意外篡改。content_length默认为None表示长度未知传输层需要按未知长度处理如分块流式传输。三、转换入口coerce_write_payload对外暴露的唯一函数是 coerce_write_payloaddef coerce_write_payload(*, path: Path, data: io.IOBase) - WritePayload: stream _BinaryReadAdapter(pathpath, streamdata) return WritePayload(streamstream, content_length_best_effort_content_length(data))它只做两件事把调用方传入的原始流data包进_BinaryReadAdapter得到保证二进制读取语义的适配流调用_best_effort_content_length(data)尝试从原始流上推断内容长度把结果随流一起封装进WritePayload。注意它必须携带path参数——该路径不参与数据转换而是用于在出错时向调用方报告是哪个目标路径的写入失败见下文错误处理小节。四、二进制读取适配器_BinaryReadAdapter_BinaryReadAdapterworkspace_payloads.py是模块中最关键的实现细节。它包装原始流向传输层暴露统一、严格的二进制读取语义class _BinaryReadAdapter(io.IOBase): def __init__(self, *, path: Path, stream: io.IOBase) - None: self._path path self._stream stream def readable(self) - bool: return True def read(self, size: int -1) - bytes: chunk self._stream.read(size) if chunk is None: return b if isinstance(chunk, bytes): return chunk if isinstance(chunk, bytearray): return bytes(chunk) raise WorkspaceWriteTypeError(pathself._path, actual_typetype(chunk).__name__) def readinto(self, b: bytearray) - int: data self.read(len(b)) n len(data) b[:n] data return n def seek(self, offset: int, whence: int io.SEEK_SET) - int: return int(self._stream.seek(offset, whence)) def tell(self) - int: return int(self._stream.tell())4.1 read()容忍 None 与 bytearray拒绝文本read()的容错与校验逻辑是整个适配器的核心read()返回None按b处理某些异步或惰性流在无数据时会返回None而非空字节串避免传输层对None做len()或拼接时报错返回bytes直接透传返回bytearray转换为bytes保证下游拿到的永远是bytes类型返回其他类型如str说明调用方传入的是文本流而非二进制流立即抛出WorkspaceWriteTypeError并把实际返回类型名写入错误上下文。4.2 readinto() / seek() / tell()补齐文件协议readinto(b)通过read(len(b))实现兼容io协议中按需读取到指定缓冲区的调用方式seek/tell直接委托底层流并强制转为int确保适配器对外承诺的类型不被底层实现破坏。这组方法的意义在于_BinaryReadAdapter实现了io.IOBase上二进制文件对象的主要协议方法因此它可以被任何期待二进制文件对象的传输层代码直接使用——无论是shutil.copyfileobj风格的分块读取还是readinto风格的零拷贝缓冲。五、内容长度推断链_best_effort_content_length_best_effort_content_length 的名字里就写着best-effort尽力而为——它按照优先级从高到低依次尝试四种途径获取长度全部失败才返回Nonedef _best_effort_content_length(stream: io.IOBase) - int | None: for attr in (content_length, length): value getattr(stream, attr, None) if isinstance(value, int) and value 0: return value headers getattr(stream, headers, None) if headers is not None: content_length None get getattr(headers, get, None) if callable(get): content_length get(Content-Length) if isinstance(content_length, str): try: parsed int(content_length) except ValueError: parsed None if parsed is not None and parsed 0: return parsed try: pos stream.tell() stream.seek(0, io.SEEK_END) end stream.tell() stream.seek(pos, io.SEEK_SET) return int(end - pos) except Exception: return None推断优先级链可概括为属性探测优先读取流的content_length或length属性兼容aiohttp等库为响应体附加长度属性的惯例要求是int且 0HTTP 头探测若流带有headers对象且其get(Content-Length)返回字符串则尝试解析为int解析失败或为负数时放弃此时不视为致命错误继续降级seek/tell 测量记录当前位置 →seek(0, SEEK_END)→tell()得到末尾位置 → 恢复原位差值即负载长度。这一招对io.BytesIO等内存流非常有效兜底None若流不可 seek如网络流、管道流导致测量抛异常则静默返回None由传输层按长度未知处理。该函数特意用try/except Exception包裹测量逻辑并返回None体现尽力而为的设计原则长度信息是优化项而非正确性前提绝不因推断失败阻塞写入流程。六、错误处理WorkspaceWriteTypeError当写入负载不是二进制文件类对象时适配器抛出 WorkspaceWriteTypeErrorclass WorkspaceWriteTypeError(WorkspaceIOError): Workspace write payload was not a binary file-like object. def __init__( self, *, path: Path, actual_type: str, context: Mapping[str, object] | None None, cause: BaseException | None None, ) - None: super().__init__( messagewrite() expects a binary file-like object, error_codeErrorCode.WORKSPACE_WRITE_TYPE_ERROR, opwrite, context{path: str(path), actual_type: actual_type, **_as_context(context)}, causecause, retryableFalse, )值得注意的工程细节错误码为ErrorCode.WORKSPACE_WRITE_TYPE_ERRORop固定为writeretryableFalse——类型错误属于调用方契约违反重试无意义上下文携带path与actual_type两个字段便于调用方精确定位哪个目标路径、传入了什么类型该异常由 _mount_security.py 纳入挂载mount安全脱敏体系与WorkspaceReadNotFoundError、WorkspaceArchiveReadError等同类工作区 IO 错误一起在序列化时会自动剔除敏感挂载凭据。七、后端调用链write() 如何消费 WritePayloadcoerce_write_payload被两个内置后端在write()实现中调用7.1 Docker 后端DockerSandboxSession.write() 的流程为async def write(self, path: Path, data: io.IOBase, *, userNone) - None: payload coerce_write_payload(pathpath, datadata) path await self._validate_path_access(path, for_writeTrue) if user is not None: await self._stream_into_exec( cmd[ sh, -lc, mkdir -p $(dirname $1) cat $1, sh, sandbox_path_str(path), ], streampayload.stream, error_pathpath, useruser, ) return parent path.parent await self.mkdir(parent, parentsTrue) # Stream into a temporary file from inside the container, then copy into place. # Avoid put_archive(): with Docker volume-driver-backed mounts attached, the daemon can # re-run volume mount setup during archive operations and some plugins reject the # duplicate Mount call for the same container id. staging_path self._archive_stage_path(name_hintpath.name) ... await self._write_stream_via_exec( staging_pathstaging_path, streampayload.stream, ... )它首先完成路径校验_validate_path_access(..., for_writeTrue)然后指定了user把payload.stream通过sh -lc mkdir -p $(dirname $1) cat $1直接流式写入目标路径未指定user先mkdir父目录再把流写入容器内暂存文件最后拷贝到目标位置。源码注释明确指出这一设计是为了规避put_archive()在挂载了 volume-driver 型卷时可能触发的插件兼容问题。7.2 Unix 本地沙箱后端UnixLocalSandboxSession.write() 同样以coerce_write_payload开头payload coerce_write_payload(pathpath, datadata) workspace_path self.normalize_path(path, for_writeTrue) if user is not None: ...两个后端统一从coerce_write_payload取得WritePayload再各自按后端特性Docker exec 流、本地文件系统把payload.stream落盘。这正体现了该模块一处规范化、多后端复用的设计价值——无论后端差异多大对调用方数据的契约校验只发生一次。八、测试验证契约行为的完整清单单元测试位于 tests/sandbox/test_workspace_payloads.py用五组用例把模块行为钉死测试验证点test_coerce_write_payload_adapts_binary_reads普通BytesIO被适配后readable()为真、read(1)/read()顺序读取正确且content_length 3test_coerce_write_payload_adapts_bytearray_and_none_readsread()返回bytearray时被转成bytes返回None时得到btest_coerce_write_payload_supports_readinto_seek_and_tellreadinto、seek、tell的语义与原生文件对象一致test_coerce_write_payload_rejects_text_chunks流返回str时抛出WorkspaceWriteTypeError错误码为WORKSPACE_WRITE_TYPE_ERROR上下文含path与actual_type: strtest_coerce_write_payload_uses_best_effort_content_length参数化验证长度推断length属性优先5Content-Length头生效7→7负数头与非法头被丢弃并回退到 seek/tell 测量均得3不可 seek 的流返回None最后一个参数化用例尤其值得细读它精确刻画了 长度推断链 的降级路径(_LengthStream(babc, 5), 5), # length 属性 → 5 (_HeaderStream(babc, 7), 7), # Content-Length 头 → 7 (_HeaderStream(babc, -1), 3), # 负数头无效 → seek 测量 → 3 (_HeaderStream(babc, invalid), 3), # 非法头 → seek 测量 → 3 (_UnseekableStream(babc), None), # 不可 seek → None九、工程启示与实践建议结合源码与测试可以总结出几条可直接指导实践的结论write()的入参必须是二进制文件类对象。传入io.StringIO或以文本模式打开的文件会触发WorkspaceWriteTypeError错误码WORKSPACE_WRITE_TYPE_ERROR。文本内容请先编码为bytes再包一层io.BytesIO。流的位置约定是起始位置。BaseSandboxSession.write()文档明确要求 data 是positioned at the start of the payload的流长度测量逻辑也会记录并恢复当前位置不会破坏调用方对流状态的预期。长度信息是优化项。若你的流带有content_length/length属性或headers[Content-Length]适配器会直接采用省去一次 seek/tell 往返不可 seek 的流也不会报错只是content_length为None。自定义流应遵循io.IOBase的读取协议保证read(size)返回bytes或bytearray/None作为边界情形并实现seek/tell至少对内存流如此即可无缝接入所有沙箱后端的write()。如果希望继续深入可以沿着以下路径阅读本仓库源码抽象契约base_sandbox_session.py 中的read/write抽象方法后端实现docker.py 与 unix_local.py错误体系errors.py 及挂载安全脱敏列表 _mount_security.py参考文档workspace_payloads.md、base_sandbox_session.md、sandbox_client.md。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考