Class AdkApp (1.118.0)

AdkApp(
    *,
    agent: BaseAgent,
    plugins: typing.Optional[typing.List[BasePlugin]] = None,
    enable_tracing: bool = False,
    session_service_builder: typing.Optional[
        typing.Callable[[...], BaseSessionService]
    ] = None,
    artifact_service_builder: typing.Optional[
        typing.Callable[[...], BaseArtifactService]
    ] = None,
    memory_service_builder: typing.Optional[
        typing.Callable[[...], BaseMemoryService]
    ] = None,
    credential_service_builder: typing.Optional[
        typing.Callable[[...], BaseCredentialService]
    ] = None,
    env_vars: typing.Optional[typing.Dict[str, str]] = None
)

An ADK Application.

Methods

AdkApp

AdkApp(
    *,
    agent: BaseAgent,
    plugins: typing.Optional[typing.List[BasePlugin]] = None,
    enable_tracing: bool = False,
    session_service_builder: typing.Optional[
        typing.Callable[[...], BaseSessionService]
    ] = None,
    artifact_service_builder: typing.Optional[
        typing.Callable[[...], BaseArtifactService]
    ] = None,
    memory_service_builder: typing.Optional[
        typing.Callable[[...], BaseMemoryService]
    ] = None,
    credential_service_builder: typing.Optional[
        typing.Callable[[...], BaseCredentialService]
    ] = None,
    env_vars: typing.Optional[typing.Dict[str, str]] = None
)

An ADK Application.

async_add_session_to_memory

async_add_session_to_memory(*, session: typing.Dict[str, typing.Any])

Generates memories.

Parameter
Name	Description
`session`	`Dict[str, Any]` Required. The session to use for generating memories. It should be a dictionary representing an ADK Session object, e.g. session.model_dump(mode="json").

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_search_memory

async_search_memory(*, user_id: str, query: str)

Searches memories for the given user.

async_stream_query

async_stream_query(
    *,
    message: typing.Union[str, typing.Dict[str, typing.Any]],
    user_id: str,
    session_id: typing.Optional[str] = None,
    run_config: typing.Optional[typing.Dict[str, typing.Any]] = 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.
`run_config`	`Optional[Dict[str, Any]]` Optional. The run config to use for the query. If you want to pass in a `run_config` pydantic object, you can pass in a dict representing it as `run_config.model_dump(mode="json")`.
`\\kwargs`	`dict[str, Any] :Yields: Event dictionaries asynchronously.` Optional. Additional keyword arguments to pass to the runner.

bidi_stream_query

bidi_stream_query(request_queue: typing.Any) -> typing.AsyncIterable[typing.Any]

Bidi streaming query the ADK application.

Exceptions
Type	Description
`TypeError`	If the request_queue is not an asyncio.Queue instance.
`ValueError`	If the first request does not have a user_id.
`ValidationError`	If failed to convert to LiveRequest. :Yields: The stream responses of querying the ADK application.

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)