Items

Item utilities for the run pipeline. Hosts input normalization helpers and lightweight builders for synthetic run items or IDs used during tool execution. Internal use only.

NestedHistoryOwnedItem dataclass

A run item and the exact nested-input occurrence that represents it.

Source code in src/agents/run_internal/items.py

@dataclass(frozen=True)
class NestedHistoryOwnedItem:
    """A run item and the exact nested-input occurrence that represents it."""

    run_item: RunItem | None
    input_index: int
    digest: str
    input_item: TResponseInputItem | None = field(default=None, compare=False, repr=False)

NestedHistoryOwnedItemRef dataclass

Durable coordinates plus the live object for one owned session occurrence.

Source code in src/agents/run_internal/items.py

@dataclass(frozen=True)
class NestedHistoryOwnedItemRef:
    """Durable coordinates plus the live object for one owned session occurrence."""

    session_index: int
    digest: str
    input_index: int
    run_item: RunItem | None = field(default=None, compare=False, repr=False)
    input_item: TResponseInputItem | None = field(default=None, compare=False, repr=False)

nested_history_run_item_occurrence_key

nested_history_run_item_occurrence_key(
    run_item: RunItem | None,
) -> str | None

Return the private copy-lineage key for a run item, when one exists.

Source code in src/agents/run_internal/items.py

def nested_history_run_item_occurrence_key(run_item: RunItem | None) -> str | None:
    """Return the private copy-lineage key for a run item, when one exists."""
    if run_item is None:
        return None
    key = getattr(run_item, _NESTED_HISTORY_RUN_ITEM_OCCURRENCE_KEY, None)
    return key if isinstance(key, str) and key else None

ensure_nested_history_run_item_occurrence_key

ensure_nested_history_run_item_occurrence_key(
    run_item: RunItem,
) -> str

Bind an ephemeral key that survives object copies but never enters model payloads.

Source code in src/agents/run_internal/items.py

def ensure_nested_history_run_item_occurrence_key(run_item: RunItem) -> str:
    """Bind an ephemeral key that survives object copies but never enters model payloads."""
    key = nested_history_run_item_occurrence_key(run_item)
    if key is None:
        key = uuid4().hex
        setattr(run_item, _NESTED_HISTORY_RUN_ITEM_OCCURRENCE_KEY, key)
    return key

copy_input_items

copy_input_items(
    value: str | list[TResponseInputItem],
) -> str | list[TResponseInputItem]

Return a shallow copy of input items so mutations do not leak between turns.

Source code in src/agents/run_internal/items.py

def copy_input_items(value: str | list[TResponseInputItem]) -> str | list[TResponseInputItem]:
    """Return a shallow copy of input items so mutations do not leak between turns."""
    return value if isinstance(value, str) else value.copy()

run_item_to_input_item

run_item_to_input_item(
    run_item: RunItem,
    reasoning_item_id_policy: ReasoningItemIdPolicy
    | None = None,
) -> TResponseInputItem | None

Convert a run item to model input, optionally stripping reasoning IDs.

Source code in src/agents/run_internal/items.py

def run_item_to_input_item(
    run_item: RunItem,
    reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
) -> TResponseInputItem | None:
    """Convert a run item to model input, optionally stripping reasoning IDs."""
    if run_item.type == "tool_approval_item":
        return None
    to_input = getattr(run_item, "to_input_item", None)
    input_item = to_input() if callable(to_input) else cast(TResponseInputItem, run_item.raw_item)
    if isinstance(input_item, dict) and input_item.get("status") is None:
        input_item = {k: v for k, v in input_item.items() if k != "status"}
    if (
        _should_omit_reasoning_item_ids(reasoning_item_id_policy)
        and run_item.type == "reasoning_item"
    ):
        return _without_reasoning_item_id(input_item)
    return cast(TResponseInputItem, input_item)

run_items_to_input_items

run_items_to_input_items(
    run_items: Sequence[RunItem],
    reasoning_item_id_policy: ReasoningItemIdPolicy
    | None = None,
) -> list[TResponseInputItem]

Convert run items to model input items while skipping approvals.

Source code in src/agents/run_internal/items.py

def run_items_to_input_items(
    run_items: Sequence[RunItem],
    reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
) -> list[TResponseInputItem]:
    """Convert run items to model input items while skipping approvals."""
    converted: list[TResponseInputItem] = []
    for run_item in run_items:
        item = run_item_to_input_item(run_item, reasoning_item_id_policy)
        if item is not None:
            converted.append(item)
    return converted

drop_orphan_function_calls

drop_orphan_function_calls(
    items: list[TResponseInputItem],
    *,
    pruning_indexes: set[int] | None = None,
) -> list[TResponseInputItem]

Remove tool and program call items that do not have corresponding outputs so resumptions or retries do not replay stale calls. Program-owned items are removed with an orphan program, while programs with retained hosted calls or tool outputs remain available for continuation. Reasoning items that immediately precede a call dropped by this pass are also removed, since the Responses API rejects reasoning items that are not followed by their associated model-emitted item (Item 'rs_...' of type 'reasoning' was provided without its required following item).

Source code in src/agents/run_internal/items.py

def drop_orphan_function_calls(
    items: list[TResponseInputItem],
    *,
    pruning_indexes: set[int] | None = None,
) -> list[TResponseInputItem]:
    """
    Remove tool and program call items that do not have corresponding outputs so resumptions or
    retries do not replay stale calls. Program-owned items are removed with an orphan program,
    while programs with retained hosted calls or tool outputs remain available for continuation.
    Reasoning items that immediately precede a call dropped by this pass are also removed, since
    the Responses API rejects reasoning items that are not followed by their associated
    model-emitted item (``Item 'rs_...' of type 'reasoning' was provided without its required
    following item``).
    """

    completed_call_ids = _completed_call_ids_by_type(items)
    matched_anonymous_tool_search_calls = _matched_anonymous_tool_search_call_indexes(items)
    active_program_call_ids: set[str] = set()
    orphan_program_call_ids: set[str] = set()

    for index, entry in enumerate(items):
        if pruning_indexes is not None and index not in pruning_indexes:
            continue
        if not isinstance(entry, dict) or entry.get("type") != "program":
            continue
        call_id = entry.get("call_id")
        if not isinstance(call_id, str):
            continue
        if call_id in completed_call_ids["program_output"]:
            continue
        if any(
            _get_program_caller_id(candidate) == call_id
            and _is_retained_program_owned_item(candidate, candidate_index, pruning_indexes)
            for candidate_index, candidate in enumerate(items)
        ):
            active_program_call_ids.add(call_id)
        else:
            orphan_program_call_ids.add(call_id)

    dropped_indexes: set[int] = set()
    filtered: list[TResponseInputItem] = []
    for index, entry in enumerate(items):
        if not isinstance(entry, dict):
            filtered.append(entry)
            continue
        entry_type = entry.get("type")
        if not isinstance(entry_type, str):
            filtered.append(entry)
            continue
        if pruning_indexes is not None and index not in pruning_indexes:
            filtered.append(entry)
            continue
        program_caller_id = _get_program_caller_id(entry)
        if program_caller_id is not None and program_caller_id in orphan_program_call_ids:
            dropped_indexes.add(index)
            continue
        output_type = _TOOL_CALL_TO_OUTPUT_TYPE.get(entry_type)
        if output_type is None:
            filtered.append(entry)
            continue
        call_id = entry.get("call_id")
        if program_caller_id is not None and _is_pending_hosted_shell_call(entry):
            filtered.append(entry)
            continue
        if entry_type == "program" and call_id in active_program_call_ids:
            filtered.append(entry)
            continue
        if isinstance(call_id, str) and call_id in completed_call_ids.get(output_type, set()):
            filtered.append(entry)
            continue
        if (
            entry_type == "tool_search_call"
            and not isinstance(call_id, str)
            and index in matched_anonymous_tool_search_calls
        ):
            filtered.append(entry)
            continue
        # Tool call entry will be dropped; record so we can also drop preceding reasoning items.
        dropped_indexes.add(index)

    if not dropped_indexes:
        return filtered
    return _drop_reasoning_items_preceding_dropped_calls(items, dropped_indexes)

ensure_input_item_format

ensure_input_item_format(
    item: TResponseInputItem,
) -> TResponseInputItem

Ensure a single item is normalized for model input.

Source code in src/agents/run_internal/items.py

def ensure_input_item_format(item: TResponseInputItem) -> TResponseInputItem:
    """Ensure a single item is normalized for model input."""
    coerced = _coerce_to_dict(item)
    if coerced is None:
        return item

    return cast(TResponseInputItem, coerced)

normalize_input_items_for_api

normalize_input_items_for_api(
    items: list[TResponseInputItem],
) -> list[TResponseInputItem]

Normalize input items for API submission.

Source code in src/agents/run_internal/items.py

def normalize_input_items_for_api(items: list[TResponseInputItem]) -> list[TResponseInputItem]:
    """Normalize input items for API submission."""

    normalized: list[TResponseInputItem] = []
    for item in items:
        coerced = _coerce_to_dict(item)
        if coerced is None:
            normalized.append(item)
            continue

        normalized_item = strip_internal_input_item_metadata(cast(TResponseInputItem, coerced))
        normalized.append(normalized_item)
    return normalized

prepare_model_input_items

prepare_model_input_items(
    caller_items: Sequence[TResponseInputItem],
    generated_items: Sequence[TResponseInputItem] = (),
) -> list[TResponseInputItem]

Normalize model input while pruning orphans only from runner-generated history.

Source code in src/agents/run_internal/items.py

def prepare_model_input_items(
    caller_items: Sequence[TResponseInputItem],
    generated_items: Sequence[TResponseInputItem] = (),
) -> list[TResponseInputItem]:
    """Normalize model input while pruning orphans only from runner-generated history."""
    normalized_caller_items = normalize_input_items_for_api(list(caller_items))
    if not generated_items:
        return normalized_caller_items

    normalized_generated_items = normalize_input_items_for_api(list(generated_items))
    filtered_generated_items = drop_orphan_function_calls(normalized_generated_items)
    return normalized_caller_items + filtered_generated_items

normalize_resumed_input

normalize_resumed_input(
    raw_input: str | list[TResponseInputItem],
) -> str | list[TResponseInputItem]

Normalize resumed list inputs and drop orphan tool calls.

Source code in src/agents/run_internal/items.py

def normalize_resumed_input(
    raw_input: str | list[TResponseInputItem],
) -> str | list[TResponseInputItem]:
    """Normalize resumed list inputs and drop orphan tool calls."""
    if isinstance(raw_input, list):
        normalized = normalize_input_items_for_api(raw_input)
        return drop_orphan_function_calls(normalized)
    return raw_input

fingerprint_input_item

fingerprint_input_item(
    item: Any, *, ignore_ids_for_matching: bool = False
) -> str | None

Hashable fingerprint used to dedupe or rewind input items across resumes.

Source code in src/agents/run_internal/items.py

def fingerprint_input_item(item: Any, *, ignore_ids_for_matching: bool = False) -> str | None:
    """Hashable fingerprint used to dedupe or rewind input items across resumes."""
    if item is None:
        return None

    try:
        payload: Any
        if hasattr(item, "model_dump"):
            payload = _model_dump_without_warnings(item)
            if payload is None:
                return None
            if isinstance(payload, dict):
                payload = cast(
                    dict[str, Any],
                    strip_internal_input_item_metadata(cast(TResponseInputItem, payload)),
                )
        elif isinstance(item, dict):
            payload = cast(
                dict[str, Any],
                strip_internal_input_item_metadata(cast(TResponseInputItem, item)),
            )
            if ignore_ids_for_matching:
                payload.pop("id", None)
        else:
            payload = ensure_input_item_format(item)
            if isinstance(payload, dict):
                payload = cast(
                    dict[str, Any],
                    strip_internal_input_item_metadata(cast(TResponseInputItem, payload)),
                )
            if ignore_ids_for_matching and isinstance(payload, dict):
                payload.pop("id", None)

        return json.dumps(payload, sort_keys=True, default=str)
    except Exception:
        return None

digest_input_item

digest_input_item(item: Any) -> str | None

Return a fixed-size digest of an input item for durable occurrence tracking.

Source code in src/agents/run_internal/items.py

def digest_input_item(item: Any) -> str | None:
    """Return a fixed-size digest of an input item for durable occurrence tracking."""
    coerced = _coerce_to_dict(item)
    if coerced is not None:
        coerced = cast(
            dict[str, Any],
            strip_internal_input_item_metadata(cast(TResponseInputItem, coerced)),
        )
        if coerced.get("role") == "assistant":
            content = coerced.get("content")
            if isinstance(content, str):
                coerced["content"] = [{"type": "output_text", "text": content}]
            if coerced.get("status") in {None, "completed"}:
                coerced.pop("status", None)
        item = coerced

    fingerprint = fingerprint_input_item(item)
    if fingerprint is None:
        return None
    return hashlib.sha256(fingerprint.encode("utf-8")).hexdigest()

filter_nested_history_owned_item_refs_for_input

filter_nested_history_owned_item_refs_for_input(
    input: str | Sequence[TResponseInputItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> list[NestedHistoryOwnedItemRef]

Keep ownership whose exact clean input occurrence is still present.

Source code in src/agents/run_internal/items.py

def filter_nested_history_owned_item_refs_for_input(
    input: str | Sequence[TResponseInputItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> list[NestedHistoryOwnedItemRef]:
    """Keep ownership whose exact clean input occurrence is still present."""
    if isinstance(input, str) or not owned_item_refs:
        return []

    input_digests = [digest_input_item(item) for item in input]
    input_indexes_by_identity: dict[tuple[int, str], deque[int]] = {}
    for index, (item, digest) in enumerate(zip(input, input_digests, strict=True)):
        if digest is not None:
            input_indexes_by_identity.setdefault((id(item), digest), deque()).append(index)

    retained: list[NestedHistoryOwnedItemRef] = []
    used_input_indexes: set[int] = set()

    def _take_unused(candidates: deque[int] | None) -> int | None:
        while candidates:
            candidate = candidates.popleft()
            if candidate not in used_input_indexes:
                return candidate
        return None

    for item_ref in owned_item_refs:
        input_index = (
            _take_unused(input_indexes_by_identity.get((id(item_ref.input_item), item_ref.digest)))
            if item_ref.input_item is not None
            else None
        )
        if input_index is None and item_ref.input_item is None:
            candidate_index = item_ref.input_index
            if (
                0 <= candidate_index < len(input)
                and candidate_index not in used_input_indexes
                and input_digests[candidate_index] == item_ref.digest
            ):
                input_index = candidate_index
        if input_index is None:
            continue
        used_input_indexes.add(input_index)
        retained.append(replace(item_ref, input_index=input_index, input_item=input[input_index]))
    return retained

reconcile_nested_history_owned_input_after_rewrite

reconcile_nested_history_owned_input_after_rewrite(
    previous_input: str | Sequence[TResponseInputItem],
    rewritten_input: str | Sequence[TResponseInputItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> tuple[
    str | list[TResponseInputItem],
    list[NestedHistoryOwnedItemRef],
]

Rebind ownership after an unambiguous input rewrite.

Source code in src/agents/run_internal/items.py

def reconcile_nested_history_owned_input_after_rewrite(
    previous_input: str | Sequence[TResponseInputItem],
    rewritten_input: str | Sequence[TResponseInputItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> tuple[str | list[TResponseInputItem], list[NestedHistoryOwnedItemRef]]:
    """Rebind ownership after an unambiguous input rewrite."""
    if isinstance(rewritten_input, str) or not owned_item_refs:
        return (
            rewritten_input if isinstance(rewritten_input, str) else list(rewritten_input),
            [],
        )
    if isinstance(previous_input, str):
        return list(rewritten_input), []

    rewritten = list(rewritten_input)
    previous = list(previous_input)
    previous_digests = [digest_input_item(item) for item in previous]
    rewritten_digests = [digest_input_item(item) for item in rewritten]
    previous_digest_counts: dict[str, int] = {}
    rewritten_digest_counts: dict[str, int] = {}
    previous_identity_digests: set[tuple[int, str]] = set()
    rewritten_indexes_by_identity: dict[tuple[int, str], deque[int]] = {}
    rewritten_indexes_by_digest: dict[str, deque[int]] = {}
    for item, digest in zip(previous, previous_digests, strict=True):
        if digest is None:
            continue
        previous_digest_counts[digest] = previous_digest_counts.get(digest, 0) + 1
        previous_identity_digests.add((id(item), digest))
    for index, (item, digest) in enumerate(zip(rewritten, rewritten_digests, strict=True)):
        if digest is None:
            continue
        rewritten_digest_counts[digest] = rewritten_digest_counts.get(digest, 0) + 1
        rewritten_indexes_by_identity.setdefault((id(item), digest), deque()).append(index)
        rewritten_indexes_by_digest.setdefault(digest, deque()).append(index)

    recoverable_ref_counts: dict[str, int] = {}
    for item_ref in owned_item_refs:
        if (
            item_ref.input_item is not None
            and (id(item_ref.input_item), item_ref.digest) in previous_identity_digests
        ):
            recoverable_ref_counts[item_ref.digest] = (
                recoverable_ref_counts.get(item_ref.digest, 0) + 1
            )
    used_indexes: set[int] = set()
    used_digest_counts: dict[str, int] = {}
    retained: list[NestedHistoryOwnedItemRef] = []

    def _take_unused(candidates: deque[int] | None) -> int | None:
        while candidates:
            candidate = candidates.popleft()
            if candidate not in used_indexes:
                return candidate
        return None

    for item_ref in owned_item_refs:
        identity_match = (
            _take_unused(
                rewritten_indexes_by_identity.get((id(item_ref.input_item), item_ref.digest))
            )
            if item_ref.input_item is not None
            else None
        )
        if identity_match is not None:
            used_indexes.add(identity_match)
            used_digest_counts[item_ref.digest] = used_digest_counts.get(item_ref.digest, 0) + 1
            retained.append(
                replace(
                    item_ref,
                    input_index=identity_match,
                    input_item=rewritten[identity_match],
                )
            )
            continue

        previous_match = (
            item_ref.input_item is not None
            and (id(item_ref.input_item), item_ref.digest) in previous_identity_digests
        )
        previous_count = previous_digest_counts.get(item_ref.digest, 0)
        rewritten_count = rewritten_digest_counts.get(item_ref.digest, 0)
        all_equal_occurrences_owned = (
            previous_count == rewritten_count == recoverable_ref_counts.get(item_ref.digest, 0)
        )
        if (
            not previous_match
            or rewritten_count <= used_digest_counts.get(item_ref.digest, 0)
            or not ((previous_count == 1 and rewritten_count == 1) or all_equal_occurrences_owned)
        ):
            continue

        candidate_index = _take_unused(rewritten_indexes_by_digest.get(item_ref.digest))
        if candidate_index is None:
            continue
        used_indexes.add(candidate_index)
        used_digest_counts[item_ref.digest] = used_digest_counts.get(item_ref.digest, 0) + 1
        retained.append(
            replace(
                item_ref,
                input_index=candidate_index,
                input_item=rewritten[candidate_index],
            )
        )

    return rewritten, retained

resolve_nested_history_owned_item_indexes

resolve_nested_history_owned_item_indexes(
    run_items: Sequence[RunItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> set[int]

Resolve ownership references without dropping a different item after list mutation.

Source code in src/agents/run_internal/items.py

def resolve_nested_history_owned_item_indexes(
    run_items: Sequence[RunItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> set[int]:
    """Resolve ownership references without dropping a different item after list mutation."""
    if not owned_item_refs:
        return set()

    indexes_by_identity_digest: dict[tuple[int, str], deque[int]] = {}
    indexes_by_occurrence_digest: dict[tuple[str, str], deque[int]] = {}
    run_item_digests: list[str | None] = []
    for index, run_item in enumerate(run_items):
        input_item = run_item_to_input_item(run_item)
        digest = digest_input_item(input_item) if input_item is not None else None
        run_item_digests.append(digest)
        if digest is None:
            continue
        indexes_by_identity_digest.setdefault((id(run_item), digest), deque()).append(index)
        occurrence_key = nested_history_run_item_occurrence_key(run_item)
        if occurrence_key is not None:
            indexes_by_occurrence_digest.setdefault((occurrence_key, digest), deque()).append(index)

    resolved: set[int] = set()

    def _peek_unused(candidates: deque[int] | None) -> int | None:
        while candidates and candidates[0] in resolved:
            candidates.popleft()
        return candidates[0] if candidates else None

    for item_ref in owned_item_refs:
        occurrence_key = nested_history_run_item_occurrence_key(item_ref.run_item)
        stored_index = item_ref.session_index
        if (
            0 <= stored_index < len(run_items)
            and stored_index not in resolved
            and run_item_digests[stored_index] == item_ref.digest
            and item_ref.run_item is not None
            and (
                run_items[stored_index] is item_ref.run_item
                or (
                    occurrence_key is not None
                    and nested_history_run_item_occurrence_key(run_items[stored_index])
                    == occurrence_key
                )
            )
        ):
            resolved.add(stored_index)
            continue

        identity_index = (
            _peek_unused(indexes_by_identity_digest.get((id(item_ref.run_item), item_ref.digest)))
            if item_ref.run_item is not None
            else None
        )
        occurrence_index = (
            _peek_unused(indexes_by_occurrence_digest.get((occurrence_key, item_ref.digest)))
            if occurrence_key is not None
            else None
        )
        candidates = [index for index in (identity_index, occurrence_index) if index is not None]
        if candidates:
            resolved.add(min(candidates))

    return resolved

rebase_nested_history_owned_item_refs

rebase_nested_history_owned_item_refs(
    input: str | Sequence[TResponseInputItem],
    run_items: Sequence[RunItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> list[NestedHistoryOwnedItemRef]

Rebase surviving ownership onto exact live input and session occurrences.

Source code in src/agents/run_internal/items.py

def rebase_nested_history_owned_item_refs(
    input: str | Sequence[TResponseInputItem],
    run_items: Sequence[RunItem],
    owned_item_refs: Sequence[NestedHistoryOwnedItemRef],
) -> list[NestedHistoryOwnedItemRef]:
    """Rebase surviving ownership onto exact live input and session occurrences."""
    retained_refs = filter_nested_history_owned_item_refs_for_input(input, owned_item_refs)
    indexes_by_identity_digest: dict[tuple[int, str], deque[int]] = {}
    indexes_by_occurrence_digest: dict[tuple[str, str], deque[int]] = {}
    for index, run_item in enumerate(run_items):
        input_item = run_item_to_input_item(run_item)
        digest = digest_input_item(input_item) if input_item is not None else None
        if digest is None:
            continue
        indexes_by_identity_digest.setdefault((id(run_item), digest), deque()).append(index)
        occurrence_key = nested_history_run_item_occurrence_key(run_item)
        if occurrence_key is not None:
            indexes_by_occurrence_digest.setdefault((occurrence_key, digest), deque()).append(index)

    rebased: list[NestedHistoryOwnedItemRef] = []
    used_indexes: set[int] = set()

    def _peek_unused(candidates: deque[int] | None) -> int | None:
        while candidates and candidates[0] in used_indexes:
            candidates.popleft()
        return candidates[0] if candidates else None

    for item_ref in retained_refs:
        occurrence_key = nested_history_run_item_occurrence_key(item_ref.run_item)
        identity_index = (
            _peek_unused(indexes_by_identity_digest.get((id(item_ref.run_item), item_ref.digest)))
            if item_ref.run_item is not None
            else None
        )
        occurrence_index = (
            _peek_unused(indexes_by_occurrence_digest.get((occurrence_key, item_ref.digest)))
            if occurrence_key is not None
            else None
        )
        candidates = [index for index in (identity_index, occurrence_index) if index is not None]
        if not candidates:
            continue
        index = min(candidates)
        used_indexes.add(index)
        rebased.append(replace(item_ref, session_index=index, run_item=run_items[index]))
    return rebased

strip_internal_input_item_metadata

strip_internal_input_item_metadata(
    item: TResponseInputItem,
) -> TResponseInputItem

Remove SDK-only session metadata before sending items back to the model.

Source code in src/agents/run_internal/items.py

def strip_internal_input_item_metadata(item: TResponseInputItem) -> TResponseInputItem:
    """Remove SDK-only session metadata before sending items back to the model."""
    if not isinstance(item, dict):
        return item

    cleaned = dict(item)
    cleaned.pop(TOOL_CALL_SESSION_DESCRIPTION_KEY, None)
    cleaned.pop(TOOL_CALL_SESSION_TITLE_KEY, None)
    return cast(TResponseInputItem, cleaned)

deduplicate_input_items

deduplicate_input_items(
    items: Sequence[TResponseInputItem],
) -> list[TResponseInputItem]

Remove duplicate items that share stable identifiers to avoid re-sending tool outputs.

Source code in src/agents/run_internal/items.py

def deduplicate_input_items(items: Sequence[TResponseInputItem]) -> list[TResponseInputItem]:
    """Remove duplicate items that share stable identifiers to avoid re-sending tool outputs."""
    seen_keys: set[str] = set()
    deduplicated: list[TResponseInputItem] = []
    for item in items:
        dedupe_key = _dedupe_key(item)
        if dedupe_key is None:
            deduplicated.append(item)
            continue
        if dedupe_key in seen_keys:
            continue
        seen_keys.add(dedupe_key)
        deduplicated.append(item)
    return deduplicated

deduplicate_input_items_preferring_latest

deduplicate_input_items_preferring_latest(
    items: Sequence[TResponseInputItem],
) -> list[TResponseInputItem]

Deduplicate by stable identifiers while keeping the latest occurrence.

Source code in src/agents/run_internal/items.py

def deduplicate_input_items_preferring_latest(
    items: Sequence[TResponseInputItem],
) -> list[TResponseInputItem]:
    """Deduplicate by stable identifiers while keeping the latest occurrence."""
    # deduplicate_input_items keeps the first item per dedupe key. Reverse twice so that
    # the latest item in the original order wins for duplicate IDs/call_ids.
    return list(reversed(deduplicate_input_items(list(reversed(items)))))

function_tool_error_output

function_tool_error_output(
    tool_call: Any,
    output: Any,
    *,
    output_json_schema: dict[str, Any] | None,
) -> Any

Encode SDK-generated programmatic tool errors as provider-compatible JSON objects.

Source code in src/agents/run_internal/items.py

def function_tool_error_output(
    tool_call: Any,
    output: Any,
    *,
    output_json_schema: dict[str, Any] | None,
) -> Any:
    """Encode SDK-generated programmatic tool errors as provider-compatible JSON objects."""
    if output_json_schema is None or not isinstance(output, str):
        return output

    if isinstance(tool_call, dict):
        caller = tool_call.get("caller")
    else:
        caller = getattr(tool_call, "caller", None)
    caller_type = caller.get("type") if isinstance(caller, dict) else getattr(caller, "type", None)
    if caller_type != "program":
        return output

    return json.dumps({"error": output}, ensure_ascii=False, separators=(",", ":"))

function_rejection_item

function_rejection_item(
    agent: Any,
    tool_call: Any,
    *,
    rejection_message: str = REJECTION_MESSAGE,
    output_json_schema: dict[str, Any] | None = None,
    scope_id: str | None = None,
    tool_origin: Any = None,
) -> ToolCallOutputItem

Build a ToolCallOutputItem representing a rejected function tool call.

Source code in src/agents/run_internal/items.py

def function_rejection_item(
    agent: Any,
    tool_call: Any,
    *,
    rejection_message: str = REJECTION_MESSAGE,
    output_json_schema: dict[str, Any] | None = None,
    scope_id: str | None = None,
    tool_origin: Any = None,
) -> ToolCallOutputItem:
    """Build a ToolCallOutputItem representing a rejected function tool call."""
    if isinstance(tool_call, ResponseFunctionToolCall):
        drop_agent_tool_run_result(tool_call, scope_id=scope_id)
    provider_output = function_tool_error_output(
        tool_call,
        rejection_message,
        output_json_schema=output_json_schema,
    )
    return ToolCallOutputItem(
        output=rejection_message,
        raw_item=ItemHelpers.tool_call_output_item(
            tool_call,
            provider_output,
        ),
        agent=agent,
        tool_origin=tool_origin,
    )

shell_rejection_item

shell_rejection_item(
    agent: Any,
    call_id: str,
    *,
    tool_call: Any | None = None,
    rejection_message: str = REJECTION_MESSAGE,
) -> ToolCallOutputItem

Build a ToolCallOutputItem representing a rejected shell call.

Source code in src/agents/run_internal/items.py

def shell_rejection_item(
    agent: Any,
    call_id: str,
    *,
    tool_call: Any | None = None,
    rejection_message: str = REJECTION_MESSAGE,
) -> ToolCallOutputItem:
    """Build a ToolCallOutputItem representing a rejected shell call."""
    rejection_output: dict[str, Any] = {
        "stdout": "",
        "stderr": rejection_message,
        "outcome": {"type": "exit", "exit_code": 1},
    }
    rejection_raw_item: dict[str, Any] = {
        "type": "shell_call_output",
        "call_id": call_id,
        "output": [rejection_output],
    }
    if tool_call is not None:
        ItemHelpers.copy_tool_call_caller(tool_call, rejection_raw_item)
    return ToolCallOutputItem(agent=agent, output=rejection_message, raw_item=rejection_raw_item)

apply_patch_rejection_item

apply_patch_rejection_item(
    agent: Any,
    call_id: str,
    *,
    tool_call: Any | None = None,
    output_type: Literal[
        "apply_patch_call_output", "custom_tool_call_output"
    ] = "apply_patch_call_output",
    rejection_message: str = REJECTION_MESSAGE,
) -> ToolCallOutputItem

Build a ToolCallOutputItem representing a rejected apply_patch call.

Source code in src/agents/run_internal/items.py

def apply_patch_rejection_item(
    agent: Any,
    call_id: str,
    *,
    tool_call: Any | None = None,
    output_type: Literal["apply_patch_call_output", "custom_tool_call_output"] = (
        "apply_patch_call_output"
    ),
    rejection_message: str = REJECTION_MESSAGE,
) -> ToolCallOutputItem:
    """Build a ToolCallOutputItem representing a rejected apply_patch call."""
    rejection_raw_item: dict[str, Any] = {
        "type": output_type,
        "call_id": call_id,
        "output": rejection_message,
    }
    if output_type == "apply_patch_call_output":
        rejection_raw_item["status"] = "failed"
    if tool_call is not None:
        ItemHelpers.copy_tool_call_caller(tool_call, rejection_raw_item)
    return ToolCallOutputItem(
        agent=agent,
        output=rejection_message,
        raw_item=rejection_raw_item,
    )

extract_mcp_request_id

extract_mcp_request_id(raw_item: Any) -> str | None

Pull the request id from hosted MCP approval payloads.

Source code in src/agents/run_internal/items.py

def extract_mcp_request_id(raw_item: Any) -> str | None:
    """Pull the request id from hosted MCP approval payloads."""
    if isinstance(raw_item, dict):
        provider_data = raw_item.get("provider_data")
        if isinstance(provider_data, dict):
            candidate = provider_data.get("id")
            if isinstance(candidate, str):
                return candidate
        candidate = raw_item.get("id") or raw_item.get("call_id")
        return candidate if isinstance(candidate, str) else None
    try:
        provider_data = getattr(raw_item, "provider_data", None)
    except Exception:
        provider_data = None
    if isinstance(provider_data, dict):
        candidate = provider_data.get("id")
        if isinstance(candidate, str):
            return candidate
    try:
        candidate = getattr(raw_item, "id", None) or getattr(raw_item, "call_id", None)
    except Exception:
        candidate = None
    return candidate if isinstance(candidate, str) else None

extract_mcp_request_id_from_run

extract_mcp_request_id_from_run(mcp_run: Any) -> str | None

Extract the hosted MCP request id from a streaming run item.

Source code in src/agents/run_internal/items.py

def extract_mcp_request_id_from_run(mcp_run: Any) -> str | None:
    """Extract the hosted MCP request id from a streaming run item."""
    request_item = getattr(mcp_run, "request_item", None) or getattr(mcp_run, "requestItem", None)
    if isinstance(request_item, dict):
        provider_data = request_item.get("provider_data")
        if isinstance(provider_data, dict):
            candidate = provider_data.get("id")
            if isinstance(candidate, str):
                return candidate
        candidate = request_item.get("id") or request_item.get("call_id")
    else:
        provider_data = getattr(request_item, "provider_data", None)
        if isinstance(provider_data, dict):
            candidate = provider_data.get("id")
            if isinstance(candidate, str):
                return candidate
        candidate = getattr(request_item, "id", None) or getattr(request_item, "call_id", None)
    return candidate if isinstance(candidate, str) else None