Class AdkApp (1.118.0)

AdkApp(
    *,
    agent: BaseAgent,
    app_name: typing.Optional[str] = None,
    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,
    instrumentor_builder: typing.Optional[typing.Callable[[...], typing.Any]] = None
)

An ADK Application.

Methods

AdkApp

AdkApp(
    *,
    agent: BaseAgent,
    app_name: typing.Optional[str] = None,
    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,
    instrumentor_builder: typing.Optional[typing.Callable[[...], typing.Any]] = None
)

An ADK Application.

See https://cloud.google.com/vertex-ai/generative-ai/docs/agent-engine/develop/adk for details on how to develop ADK applications on Agent Engine.

Parameters
Name	Description
`agent`	`google.adk.agents.BaseAgent` Required. The ADK agent to run.
`app_name`	`str` Optional. The name of the ADK application. Defaults to "default-app-name" when running locally, and to the corresponding agent engine ID when deployed on Agent Engine.
`plugins`	`List[BasePlugin]` Optional. The plugins to use for the ADK application. Defaults to an empty list.
`enable_tracing`	`bool` Optional. Whether to enable tracing in Cloud Trace. Defaults to False.
`session_service_builder`	`Callable[..., BaseSessionService]` Optional. A callable that returns an ADK session service. Defaults to a callable that returns InMemorySessionService when running locally and VertexAiSessionService when running on Agent Engine.
`artifact_service_builder`	`Callable[..., BaseArtifactService]` Optional. A callable that returns an ADK artifact service. Defaults to a callable that returns InMemoryArtifactService.
`memory_service_builder`	`Callable[..., BaseMemoryService]` Optional. A callable that returns an ADK memory service. Defaults to a callable that returns InMemoryMemoryService when running locally and VertexAiMemoryBankService when running on Agent Engine.
`instrumentor_builder`	`Callable[..., Any]` Optional. Callable that returns a new instrumentor. This can be used for customizing the instrumentation logic of the Agent. If not provided, a default instrumentor builder will be used. This parameter is ignored if `enable_tracing` is False.

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.

clone

clone()

Returns a clone of the ADK application.

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.

streaming_agent_run_with_events

streaming_agent_run_with_events(request_json: str)