feat: Add Google Gemini plugin for Flyte

[andreahlert]

Summary

This PR adds a new plugin for integrating Google's Gemini API with Flyte, following the same patterns as the existing OpenAI and Anthropic plugins.

Features

• function_tool decorator: Convert Flyte tasks to Gemini function declarations automatically
• Automatic type conversion: Python type hints to JSON schema for tool input validation
• Agent class: Configure Gemini agents with model, instructions, and tools
• run_agent function: Execute agent loops that handle function calling and responses
• Support for sync and async tasks
• Manual function calling control: Uses AFC (Automatic Function Calling) disabled mode for explicit tool execution management

Usage Example

import flyte
from flyteplugins.gemini import function_tool, run_agent

env = flyte.TaskEnvironment("gemini-agent")

@env.task
async def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Weather in {city}: sunny, 72F"

@env.task
async def agent_task(prompt: str) -> str:
    tools = [function_tool(get_weather)]
    return await run_agent(
        prompt=prompt,
        tools=tools,
        model="gemini-2.5-flash",
    )

Files Added

• plugins/gemini/ - Plugin package
  • pyproject.toml - Package configuration
  • README.md - Documentation
  • src/flyteplugins/gemini/agents/_function_tools.py - Core implementation
  • tests/test_agents.py - Test suite (19 tests)
• examples/genai/gemini_agent.py - Example workflow

Test Plan

• All 19 unit tests pass
• Type conversion tests (string, int, float, bool, list, dict, Optional)
• Function schema extraction tests
• FunctionTool creation tests (regular functions, Flyte tasks, async)
• Agent configuration tests
• Tool execution tests (sync and async)
• run_agent tests (tool calls, errors, max iterations, finish reasons)

Why Gemini?

Google Gemini is one of the leading LLM providers alongside OpenAI and Anthropic. Having first-class support enables users to:

• Choose the best model for their use case
• Easily switch between providers
• Use Gemini's unique capabilities (multimodal, large context windows, cost efficiency)

This plugin mirrors the Anthropic plugin's API design, making it easy to switch between providers.

[kumare3]

you should not need to do this, this is already done as part of the AsyncFunctionTemplate - native interface

[andreahlert]

I checked NativeInterface before going this route. It stores Python types as Dict[str, Tuple[Type, Any]] but doesn't expose a JSON schema conversion. Gemini's FunctionDeclaration needs parameters_json_schema as a JSON schema dict, so some conversion layer is needed regardless.

This is the same approach used in the Anthropic plugin (plugins/anthropic/.../agents/_function_tools.py, lines 34-66). Also, function_tool() needs to support regular callables and @flyte.trace decorated functions too, where NativeInterface isn't available.

Happy to refactor if there's a method I'm missing that already does this conversion.

[kumare3]

you already have native_interface

[andreahlert]

native_interface is only set when func is an AsyncFunctionTaskTemplate. For plain callables it's None, so we'd still need the inspect fallback. The Anthropic plugin stores native_interface the same way but generates the schema from the raw function for this reason.

One thing I could do is use native_interface.inputs when available and fall back to inspect otherwise. Would that be the direction you'd prefer?

[kumare3]

this is not needed and dangerous. this will start non daemon threads which will block. you dont need to do it, this is all solved by flyte, it gives you an async interface. doing the to_thread is not recommended.

i think we should just call the async execute method. it takes care of the rest

[andreahlert]

I hear you on avoiding unnecessary threads. I looked at how the SDK handles this internally, and AsyncFunctionTaskTemplate.execute() itself uses run_sync_with_loop() for sync functions, which also spins up a thread with its own event loop (src/flyte/_utils/asyncify.py).

The reason for to_thread here: for sync Flyte tasks, task(**kwargs) goes through __call__ -> controller.submit_sync() which returns a blocking Future.result(None). Without to_thread, that blocks the event loop while the agent loop needs to stay async to handle multiple tool calls.

That said, if there's a cleaner async path I should be using instead, I'm happy to switch. The Anthropic plugin uses the same to_thread pattern, so if there's a better approach we could align both.

[kumare3]

I dont think this is right, i have not looked at the anthropic plugin, but this will result in bugs and instability