Class AdkApp (1.95.1) 

AdkApp(
    *,
    agent: BaseAgent,
    enable_tracing: bool = False,
    session_service_builder: typing.Optional[
        typing.Callable[[...], BaseSessionService]
    ] = None,
    artifact_service_builder: typing.Optional[
        typing.Callable[[...], BaseArtifactService]
    ] = None,
    env_vars: typing.Optional[typing.Dict[str, str]] = None
)

An ADK Application.

Methods

AdkApp

AdkApp(
    *,
    agent: BaseAgent,
    enable_tracing: bool = False,
    session_service_builder: typing.Optional[
        typing.Callable[[...], BaseSessionService]
    ] = None,
    artifact_service_builder: typing.Optional[
        typing.Callable[[...], BaseArtifactService]
    ] = None,
    env_vars: typing.Optional[typing.Dict[str, str]] = None
)

An ADK Application.

async_create_session

async_create_session(
    *,
    user_id: str,
    session_id: typing.Optional[str] = None,
    state: typing.Optional[typing.Dict[str, typing.Any]] = None,
    **kwargs
)

Creates a new session.

Parameters
Name Description
user_id str

Required. The ID of the user.
session_id str

Optional. The ID of the session. If not provided, an ID will be be generated for the session.
state dict[str, Any]

Optional. The initial state of the session.
\*\*kwargs dict[str, Any]

Optional. Additional keyword arguments to pass to the session service.
Returns
Type Description
Session The newly created session instance.

async_delete_session

async_delete_session(*, user_id: str, session_id: str, **kwargs)

Deletes a session for the given user.

Parameters
Name Description
user_id str

Required. The ID of the user.
session_id str

Required. The ID of the session.
\*\*kwargs dict[str, Any]

Optional. Additional keyword arguments to pass to the session service.

async_get_session

async_get_session(*, user_id: str, session_id: str, **kwargs)

Get a session for the given user.

Parameters
Name Description
user_id str

Required. The ID of the user.
session_id str

Required. The ID of the session.
\*\*kwargs dict[str, Any]

Optional. Additional keyword arguments to pass to the session service.
Exceptions
Type Description
RuntimeError If the session is not found.
Returns
Type Description
Session The session instance (if any). It returns None if the session is not found.

async_list_sessions

async_list_sessions(*, user_id: str, **kwargs)

List sessions for the given user.

Parameters
Name Description
user_id str

Required. The ID of the user.
\*\*kwargs dict[str, Any]

Optional. Additional keyword arguments to pass to the session service.
Returns
Type Description
ListSessionsResponse The list of sessions.

async_stream_query

async_stream_query(
    *,
    message: typing.Union[str, typing.Dict[str, typing.Any]],
    user_id: str,
    session_id: typing.Optional[str] = None,
    **kwargs
) -> typing.AsyncIterable[typing.Dict[str, typing.Any]]

Streams responses asynchronously from the ADK application.

Parameters
Name Description
message str

Required. The message to stream responses for.
user_id str

Required. The ID of the user.
session_id str

Optional. The ID of the session. If not provided, a new session will be created for the user.
\*\*kwargs dict[str, Any] :Yields: Event dictionaries asynchronously.

Optional. Additional keyword arguments to pass to the runner.

clone

clone()

Returns a clone of the ADK application.

create_session

create_session(
    *,
    user_id: str,
    session_id: typing.Optional[str] = None,
    state: typing.Optional[typing.Dict[str, typing.Any]] = None,
    **kwargs
)

Creates a new session.

delete_session

delete_session(*, user_id: str, session_id: str, **kwargs)

Deletes a session for the given user.

get_session

get_session(*, user_id: str, session_id: str, **kwargs)

Get a session for the given user.

list_sessions

list_sessions(*, user_id: str, **kwargs)

List sessions for the given user.

register_operations

register_operations() -> typing.Dict[str, typing.List[str]]

Registers the operations of the ADK application.

set_up

set_up()

Sets up the ADK application.

stream_query

stream_query(
    *,
    message: typing.Union[str, typing.Dict[str, typing.Any]],
    user_id: str,
    session_id: typing.Optional[str] = None,
    **kwargs
)

Streams responses from the ADK application in response to a message.

Parameters
Name Description
message Union[str, Dict[str, Any]]

Required. The message to stream responses for.
user_id str

Required. The ID of the user.
session_id str

Optional. The ID of the session. If not provided, a new session will be created for the user.
\*\*kwargs dict[str, Any] :Yields: The output of querying the ADK application.

Optional. Additional keyword arguments to pass to the runner.