File size: 3,122 Bytes
5f25733 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from .generated.v2_all import FunctionCallOutputContentItem, TurnToolOutput
from .models import JsonObject
@dataclass(slots=True)
class TextInput:
"""Text supplied to a turn or steering request."""
text: str
@dataclass(slots=True)
class ImageInput:
"""Image data URL supplied as turn input."""
url: str
@dataclass(slots=True)
class LocalImageInput:
"""Local image path supplied as turn input."""
path: str
@dataclass(slots=True)
class SkillInput:
"""Named skill reference supplied as turn input."""
name: str
path: str
@dataclass(slots=True)
class MentionInput:
"""Named resource mention supplied as turn input."""
name: str
path: str
@dataclass(slots=True)
class ExternalMessage:
"""Untrusted content supplied by another agent, tool, or application.
Content has tool-level authority, below user and developer instructions. It
does not establish user authorization or approval. Pass this as the whole
input to ``thread.run()`` or ``thread.turn()`` to start a turn or join an
active regular turn. ``tool_name`` identifies the tool delivering it.
``content`` accepts text or Responses-compatible function-output content
items. Structured items can be dictionaries; no generated wrapper is needed.
"""
tool_name: str
content: str | Sequence[JsonObject | FunctionCallOutputContentItem]
namespace: str | None = None
InputItem = TextInput | ImageInput | LocalImageInput | SkillInput | MentionInput
Input = list[InputItem] | InputItem
RunInput = Input | str | ExternalMessage
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: Input | str) -> Input:
if isinstance(input, str):
return TextInput(input)
return input
def _to_wire_turn_input(input: RunInput) -> tuple[list[JsonObject], TurnToolOutput | None]:
if isinstance(input, ExternalMessage):
if not isinstance(input.tool_name, str) or not input.tool_name.strip():
raise ValueError("ExternalMessage.tool_name must be a nonempty string")
return [], TurnToolOutput.model_validate(
{"name": input.tool_name, "namespace": input.namespace, "output": input.content}
)
return _to_wire_input(_normalize_run_input(input)), None
|