Add Python SDK thread.run convenience methods (#15088)

## TL;DR
Add `thread.run(...)` / `async thread.run(...)` convenience methods to
the Python SDK for the common case.

- add `RunInput = Input | str` and `RunResult` with `final_response`,
collected `items`, and optional `usage`
- keep `thread.turn(...)` strict and lower-level for streaming,
steering, interrupting, and raw generated `Turn` access
- update Python SDK docs, quickstart examples, and tests for the sync
and async convenience flows

## Validation
- `python3 -m pytest sdk/python/tests/test_public_api_signatures.py
sdk/python/tests/test_public_api_runtime_behavior.py`
- `python3 -m pytest
sdk/python/tests/test_real_app_server_integration.py -k
'thread_run_convenience or async_thread_run_convenience'` (skipped in
this environment)

---------

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Shaqayeq
2026-03-18 17:57:48 -07:00
committed by GitHub
parent 825d09373d
commit 4fd2774614
12 changed files with 760 additions and 103 deletions

View File

@@ -0,0 +1,63 @@
from __future__ import annotations
from dataclasses import dataclass
from .models import JsonObject
@dataclass(slots=True)
class TextInput:
text: str
@dataclass(slots=True)
class ImageInput:
url: str
@dataclass(slots=True)
class LocalImageInput:
path: str
@dataclass(slots=True)
class SkillInput:
name: str
path: str
@dataclass(slots=True)
class MentionInput:
name: str
path: str
InputItem = TextInput | ImageInput | LocalImageInput | SkillInput | MentionInput
Input = list[InputItem] | InputItem
RunInput = Input | str
def _to_wire_item(item: InputItem) -> JsonObject:
if isinstance(item, TextInput):
return {"type": "text", "text": item.text}
if isinstance(item, ImageInput):
return {"type": "image", "url": item.url}
if isinstance(item, LocalImageInput):
return {"type": "localImage", "path": item.path}
if isinstance(item, SkillInput):
return {"type": "skill", "name": item.name, "path": item.path}
if isinstance(item, MentionInput):
return {"type": "mention", "name": item.name, "path": item.path}
raise TypeError(f"unsupported input item: {type(item)!r}")
def _to_wire_input(input: Input) -> list[JsonObject]:
if isinstance(input, list):
return [_to_wire_item(i) for i in input]
return [_to_wire_item(input)]
def _normalize_run_input(input: RunInput) -> Input:
if isinstance(input, str):
return TextInput(input)
return input