Google ADK

Google ADK (Agent Development Kit) adapter for Flyte.

Bring your own google-adk agent and run it durably on Flyte. ADK’s Runner owns the loop; Flyte is the runtime underneath: the tools you expose are Flyte tasks (durable child actions), each model turn is recorded for replay (durable=True), the run timeline renders into the task report, and memory_key gives cross-run conversation memory.

• flyteplugins.agents.google.tool — turn an @env.task into a Google ADK tool.
• flyteplugins.agents.google.run_agent — run the ADK agent loop inside your task and return the answer.
• flyteplugins.agents.google.durable_model — wrap a model so its turns are durable, for hand-built agent trees (e.g. sub-agent transfers) passed to run_agent via agent=.

Set the model provider’s API key in the environment (e.g. GOOGLE_API_KEY for Gemini) — wire it as a Flyte secret.

Directory

Classes

Methods

Methods

durable_model()

def durable_model(
    model: typing.Any,
) -> typing.Any

Wrap model (a name string or BaseLlm) so its turns are durable.

Returns a flyteplugins.agents.google.FlyteLlm over the resolved inner model, or model unchanged when it can’t be wrapped (durability is best-effort, never fatal).

run_agent()

def run_agent(
    input: str,
    agent: typing.Any = None,
    tools: typing.Sequence[typing.Any] = (),
    model: str = 'gemini-2.0-flash',
    instructions: str | None = None,
    name: str = 'assistant',
    max_llm_calls: int | None = None,
    durable: bool = True,
    observability: bool = True,
    memory_key: str | None = None,
    app_name: str = 'flyte-agent',
    user_id: str = 'flyte-user',
) -> str

Run a Google ADK agent with the given tools and prompt; return the final text.

Await this from an async task as await run_agent(...); from a sync task use flyteplugins.agents.google.run_agent_sync instead.

Call this from inside an @env.task — that task is the durable parent, and each tool the agent calls runs as a durable Flyte child action. Provide either a pre-built agent (an ADK LlmAgent/BaseAgent) or tools + model + instructions to have one built.

run_agent_sync()

def run_agent_sync(
    input: str,
    agent: typing.Any = None,
    tools: typing.Sequence[typing.Any] = (),
    model: str = 'gemini-2.0-flash',
    instructions: str | None = None,
    name: str = 'assistant',
    max_llm_calls: int | None = None,
    durable: bool = True,
    observability: bool = True,
    memory_key: str | None = None,
    app_name: str = 'flyte-agent',
    user_id: str = 'flyte-user',
) -> str

Synchronous variant of run_agent for use in sync tasks; runs the async implementation on a dedicated event loop.

Run a Google ADK agent with the given tools and prompt; return the final text.

Await this from an async task as await run_agent(...); from a sync task use flyteplugins.agents.google.run_agent_sync instead.

Call this from inside an @env.task — that task is the durable parent, and each tool the agent calls runs as a durable Flyte child action. Provide either a pre-built agent (an ADK LlmAgent/BaseAgent) or tools + model + instructions to have one built.

tool()

def tool(
    func: AsyncFunctionTaskTemplate | typing.Callable | None = None,
    name: str | None = None,
    description: str | None = None,
) -> typing.Callable

Wrap a Flyte @env.task as a plain async tool function — the generic default.

For SDKs that accept plain Python callables as tools (deriving the schema from the signature + docstring), this is the whole adapter tool: the returned function carries the task’s signature (functools.wraps), dispatches to task.aio() (so each call is a durable Flyte child action), exposes __wrapped_task__, and wires the backing task to flyteplugins.agents.core.ToolTaskResolver. Adapters whose SDK needs a native tool type (e.g. OpenAI’s FunctionTool, Claude’s MCP SdkMcpTool) provide their own instead.

Also accepts any other callable — a plain function or an instance of a callable class defining __call__ — and returns it usable as a tool as-is, since the plain-callable SDKs derive the schema by inspecting the callable (a class instance is inspected through its __call__). A name or description override is applied to the callable best-effort.

Usable bare, parametrized or as a direct call:

@tool
@env.task
async def get_weather(city: str) -> str: ...