Session

Session

Bases: Protocol

Protocol for session implementations.

Session stores conversation history for a specific session, allowing agents to maintain context without requiring explicit manual memory management.

ソースコード位置： src/agents/memory/session.py

@runtime_checkable
class Session(Protocol):
    """Protocol for session implementations.

    Session stores conversation history for a specific session, allowing
    agents to maintain context without requiring explicit manual memory management.
    """

    session_id: str
    session_settings: SessionSettings | None = None

    async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
        """Retrieve the conversation history for this session.

        Args:
            limit: Maximum number of items to retrieve. If None, retrieves all items.
                   When specified, returns the latest N items in chronological order.

        Returns:
            List of input items representing the conversation history
        """
        ...

    async def add_items(self, items: list[TResponseInputItem]) -> None:
        """Add new items to the conversation history.

        Args:
            items: List of input items to add to the history
        """
        ...

    async def pop_item(self) -> TResponseInputItem | None:
        """Remove and return the most recent item from the session.

        Returns:
            The most recent item if it exists, None if the session is empty
        """
        ...

    async def clear_session(self) -> None:
        """Clear all items for this session."""
        ...

get_items async

get_items(
    limit: int | None = None,
) -> list[TResponseInputItem]

Retrieve the conversation history for this session.

引数：

名前	タイプ	デスクリプション	デフォルト
limit	int | None	Maximum number of items to retrieve. If None, retrieves all items. When specified, returns the latest N items in chronological order.	None

戻り値：

タイプ	デスクリプション
list[TResponseInputItem]	List of input items representing the conversation history

ソースコード位置： src/agents/memory/session.py

async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
    """Retrieve the conversation history for this session.

    Args:
        limit: Maximum number of items to retrieve. If None, retrieves all items.
               When specified, returns the latest N items in chronological order.

    Returns:
        List of input items representing the conversation history
    """
    ...

add_items async

add_items(items: list[TResponseInputItem]) -> None

Add new items to the conversation history.

引数：

名前	タイプ	デスクリプション	デフォルト
items	list[TResponseInputItem]	List of input items to add to the history	必須

ソースコード位置： src/agents/memory/session.py

async def add_items(self, items: list[TResponseInputItem]) -> None:
    """Add new items to the conversation history.

    Args:
        items: List of input items to add to the history
    """
    ...

pop_item async

pop_item() -> TResponseInputItem | None

Remove and return the most recent item from the session.

戻り値：

タイプ	デスクリプション
TResponseInputItem | None	The most recent item if it exists, None if the session is empty

ソースコード位置： src/agents/memory/session.py

async def pop_item(self) -> TResponseInputItem | None:
    """Remove and return the most recent item from the session.

    Returns:
        The most recent item if it exists, None if the session is empty
    """
    ...

clear_session async

clear_session() -> None

Clear all items for this session.

ソースコード位置： src/agents/memory/session.py

async def clear_session(self) -> None:
    """Clear all items for this session."""
    ...

SessionABC

Bases: ABC

Abstract base class for session implementations.

Session stores conversation history for a specific session, allowing agents to maintain context without requiring explicit manual memory management.

This ABC is intended for internal use and as a base class for concrete implementations. Third-party libraries should implement the Session protocol instead.

ソースコード位置： src/agents/memory/session.py

class SessionABC(ABC):
    """Abstract base class for session implementations.

    Session stores conversation history for a specific session, allowing
    agents to maintain context without requiring explicit manual memory management.

    This ABC is intended for internal use and as a base class for concrete implementations.
    Third-party libraries should implement the Session protocol instead.
    """

    session_id: str
    session_settings: SessionSettings | None = None

    @abstractmethod
    async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
        """Retrieve the conversation history for this session.

        Args:
            limit: Maximum number of items to retrieve. If None, retrieves all items.
                   When specified, returns the latest N items in chronological order.

        Returns:
            List of input items representing the conversation history
        """
        ...

    @abstractmethod
    async def add_items(self, items: list[TResponseInputItem]) -> None:
        """Add new items to the conversation history.

        Args:
            items: List of input items to add to the history
        """
        ...

    @abstractmethod
    async def pop_item(self) -> TResponseInputItem | None:
        """Remove and return the most recent item from the session.

        Returns:
            The most recent item if it exists, None if the session is empty
        """
        ...

    @abstractmethod
    async def clear_session(self) -> None:
        """Clear all items for this session."""
        ...

get_items abstractmethod async

get_items(
    limit: int | None = None,
) -> list[TResponseInputItem]

Retrieve the conversation history for this session.

引数：

名前	タイプ	デスクリプション	デフォルト
limit	int | None	Maximum number of items to retrieve. If None, retrieves all items. When specified, returns the latest N items in chronological order.	None

戻り値：

タイプ	デスクリプション
list[TResponseInputItem]	List of input items representing the conversation history

ソースコード位置： src/agents/memory/session.py

@abstractmethod
async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
    """Retrieve the conversation history for this session.

    Args:
        limit: Maximum number of items to retrieve. If None, retrieves all items.
               When specified, returns the latest N items in chronological order.

    Returns:
        List of input items representing the conversation history
    """
    ...

add_items abstractmethod async

add_items(items: list[TResponseInputItem]) -> None

Add new items to the conversation history.

引数：

名前	タイプ	デスクリプション	デフォルト
items	list[TResponseInputItem]	List of input items to add to the history	必須

ソースコード位置： src/agents/memory/session.py

@abstractmethod
async def add_items(self, items: list[TResponseInputItem]) -> None:
    """Add new items to the conversation history.

    Args:
        items: List of input items to add to the history
    """
    ...

pop_item abstractmethod async

pop_item() -> TResponseInputItem | None

Remove and return the most recent item from the session.

戻り値：

タイプ	デスクリプション
TResponseInputItem | None	The most recent item if it exists, None if the session is empty

ソースコード位置： src/agents/memory/session.py

@abstractmethod
async def pop_item(self) -> TResponseInputItem | None:
    """Remove and return the most recent item from the session.

    Returns:
        The most recent item if it exists, None if the session is empty
    """
    ...

clear_session abstractmethod async

clear_session() -> None

Clear all items for this session.

ソースコード位置： src/agents/memory/session.py

@abstractmethod
async def clear_session(self) -> None:
    """Clear all items for this session."""
    ...

OpenAIResponsesCompactionArgs

Bases: TypedDict

Arguments for the run_compaction method.

ソースコード位置： src/agents/memory/session.py

class OpenAIResponsesCompactionArgs(TypedDict, total=False):
    """Arguments for the run_compaction method."""

    response_id: str
    """The ID of the last response to use for compaction."""

    compaction_mode: Literal["previous_response_id", "input", "auto"]
    """How to provide history for compaction.

    - "auto": Use input when the last response was not stored or no response ID is available.
    - "previous_response_id": Use server-managed response history.
    - "input": Send locally stored session items as input.
    """

    store: bool
    """Whether the last model response was stored on the server.

    When set to False, compaction should avoid "previous_response_id" unless explicitly requested.
    """

    force: bool
    """Whether to force compaction even if the threshold is not met."""

response_id instance-attribute

response_id: str

The ID of the last response to use for compaction.

compaction_mode instance-attribute

compaction_mode: Literal[
    "previous_response_id", "input", "auto"
]

How to provide history for compaction.

• "auto": Use input when the last response was not stored or no response ID is available.
• "previous_response_id": Use server-managed response history.
• "input": Send locally stored session items as input.

store instance-attribute

store: bool

Whether the last model response was stored on the server.

When set to False, compaction should avoid "previous_response_id" unless explicitly requested.

force instance-attribute

force: bool

Whether to force compaction even if the threshold is not met.

OpenAIResponsesCompactionAwareSession

Bases: Session, Protocol

Protocol for session implementations that support responses compaction.

ソースコード位置： src/agents/memory/session.py

@runtime_checkable
class OpenAIResponsesCompactionAwareSession(Session, Protocol):
    """Protocol for session implementations that support responses compaction."""

    async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None) -> None:
        """Run the compaction process for the session."""
        ...

run_compaction async

run_compaction(
    args: OpenAIResponsesCompactionArgs | None = None,
) -> None

Run the compaction process for the session.

ソースコード位置： src/agents/memory/session.py

async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None) -> None:
    """Run the compaction process for the session."""
    ...

get_items async

get_items(
    limit: int | None = None,
) -> list[TResponseInputItem]

Retrieve the conversation history for this session.

引数：

名前	タイプ	デスクリプション	デフォルト
limit	int | None	Maximum number of items to retrieve. If None, retrieves all items. When specified, returns the latest N items in chronological order.	None

戻り値：

タイプ	デスクリプション
list[TResponseInputItem]	List of input items representing the conversation history

ソースコード位置： src/agents/memory/session.py

async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
    """Retrieve the conversation history for this session.

    Args:
        limit: Maximum number of items to retrieve. If None, retrieves all items.
               When specified, returns the latest N items in chronological order.

    Returns:
        List of input items representing the conversation history
    """
    ...

add_items async

add_items(items: list[TResponseInputItem]) -> None

Add new items to the conversation history.

引数：

名前	タイプ	デスクリプション	デフォルト
items	list[TResponseInputItem]	List of input items to add to the history	必須

ソースコード位置： src/agents/memory/session.py

async def add_items(self, items: list[TResponseInputItem]) -> None:
    """Add new items to the conversation history.

    Args:
        items: List of input items to add to the history
    """
    ...

pop_item async

pop_item() -> TResponseInputItem | None

Remove and return the most recent item from the session.

戻り値：

タイプ	デスクリプション
TResponseInputItem | None	The most recent item if it exists, None if the session is empty

ソースコード位置： src/agents/memory/session.py

async def pop_item(self) -> TResponseInputItem | None:
    """Remove and return the most recent item from the session.

    Returns:
        The most recent item if it exists, None if the session is empty
    """
    ...

clear_session async

clear_session() -> None

Clear all items for this session.

ソースコード位置： src/agents/memory/session.py

async def clear_session(self) -> None:
    """Clear all items for this session."""
    ...

await_session_mutation async

await_session_mutation(awaitable: Awaitable[_T]) -> _T

Wait for a session mutation to settle before propagating caller cancellation.

Once this coroutine starts, cancellation of its caller does not cancel the supplied awaitable. Repeated cancellation requests are deferred until the operation finishes. Without caller cancellation, return the result or raise the operation's exception. With caller cancellation, retrieve the operation's outcome and raise the first caller cancellation instead, even if the operation failed. The caller must not interpret that cancellation as a rolled-back write.

This helper does not provide a timeout, transaction, retry, or mutual exclusion. A stuck operation can delay cancellation indefinitely; configure timeouts on the underlying client. Independently cancelling the supplied operation can still interrupt it. Cancellation before this coroutine starts is not covered.

ソースコード位置： src/agents/memory/session.py

async def await_session_mutation(awaitable: Awaitable[_T]) -> _T:
    """Wait for a session mutation to settle before propagating caller cancellation.

    Once this coroutine starts, cancellation of its caller does not cancel the
    supplied awaitable. Repeated cancellation requests are deferred until the
    operation finishes. Without caller cancellation, return the result or raise
    the operation's exception. With caller cancellation, retrieve the operation's
    outcome and raise the first caller cancellation instead, even if the operation
    failed. The caller must not interpret that cancellation as a rolled-back write.

    This helper does not provide a timeout, transaction, retry, or mutual exclusion.
    A stuck operation can delay cancellation indefinitely; configure timeouts on
    the underlying client. Independently cancelling the supplied operation can
    still interrupt it. Cancellation before this coroutine starts is not covered.
    """
    task = asyncio.ensure_future(awaitable)
    cancellation: asyncio.CancelledError | None = None
    while not task.done():
        try:
            await asyncio.wait({task})
        except asyncio.CancelledError as exc:
            if cancellation is None:
                cancellation = exc

    try:
        result = task.result()
    except BaseException:
        if cancellation is not None:
            raise cancellation from None
        raise
    if cancellation is not None:
        raise cancellation from None
    return result

is_openai_responses_compaction_aware_session

is_openai_responses_compaction_aware_session(
    session: Session | None,
) -> TypeGuard[OpenAIResponsesCompactionAwareSession]

Check if a session supports responses compaction.

ソースコード位置： src/agents/memory/session.py

def is_openai_responses_compaction_aware_session(
    session: Session | None,
) -> TypeGuard[OpenAIResponsesCompactionAwareSession]:
    """Check if a session supports responses compaction."""
    if session is None:
        return False
    try:
        run_compaction = getattr(session, "run_compaction", None)
    except Exception:
        return False
    return callable(run_compaction)