Skip to content

Gemini Adapters

Gemini SDK agent adapters — wraps a google-genai Client as an Agent.

gemini

Gemini SDK agent adapters — wraps a google-genai Client as an Agent.

Provides two adapters:

  • GeminiAgent: standard generate_content (request/response)
  • GeminiLiveAgent: Live API over WebSocket (streaming/real-time)

Send audio directly to a Gemini model via the SDK, no HTTP endpoint needed. Requires the google-genai package (already a core dependency).

GeminiAgent 

GeminiAgent(*, client: Any | None = None, model: str = 'gemini-2.0-flash', tools: list[Any] | None = None, system_instruction: str | None = None, config: Any | None = None)

Agent adapter that wraps a google.genai.Client object directly.

Sends audio to Gemini via client.aio.models.generate_content() and auto-parses function-call responses with :class:GeminiResponseParser.

Usage::

from google import genai

client = genai.Client(api_key="...")
agent = GeminiAgent(
    client=client,
    model="gemini-2.0-flash",
    tools=[book_flight_declaration],
)
response = await agent.run(audio)

For Vertex AI::

client = genai.Client(vertexai=True, project="my-project", location="us-central1")
agent = GeminiAgent(client=client, model="gemini-2.0-flash", tools=[...])
Source code in src/russo/adapters/gemini.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    *,
    client: Any | None = None,
    model: str = "gemini-2.0-flash",
    tools: list[Any] | None = None,
    system_instruction: str | None = None,
    config: Any | None = None,
) -> None:
    self.client = client if client is not None else _make_client()
    self.model = model
    self.tools = tools
    self.system_instruction = system_instruction
    self.config = config
    self._parser = GeminiResponseParser()

run async 

run(audio: Audio) -> AgentResponse

Send audio to Gemini and parse the tool-call response.

Source code in src/russo/adapters/gemini.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def run(self, audio: Audio) -> AgentResponse:
    """Send audio to Gemini and parse the tool-call response."""
    from google.genai import types

    data, mime_type = AudioManager.prepare_for_generate_content(audio)
    contents = [types.Part.from_bytes(data=data, mime_type=mime_type)]
    config = self._build_config(types)

    logger.debug("Sending %d bytes of %s audio to %s", len(data), mime_type, self.model)

    response = await self.client.aio.models.generate_content(
        model=self.model,
        contents=contents,
        config=config,
    )
    return self._parser.parse(response)

GeminiLiveAgent 

GeminiLiveAgent(*, client: Any | None = None, session: Any | None = None, model: str = 'gemini-live-2.5-flash-native-audio', tools: list[Any] | None = None, system_instruction: str | None = None, config: Any | None = None, response_timeout: float = 30.0)

Agent adapter for Gemini's Live API (streaming/real-time).

Connects via client.aio.live.connect(), sends audio via send_realtime_input, and collects function-call responses.

Accepts either a google.genai.Client (new session per run) or a pre-existing Live session.

Source code in src/russo/adapters/gemini.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def __init__(
    self,
    *,
    client: Any | None = None,
    session: Any | None = None,
    model: str = "gemini-live-2.5-flash-native-audio",
    tools: list[Any] | None = None,
    system_instruction: str | None = None,
    config: Any | None = None,
    response_timeout: float = 30.0,
) -> None:
    if session is None and client is None:
        client = _make_client()
    self.client = client
    self.session = session
    self.model = model
    self.tools = tools
    self.system_instruction = system_instruction
    self.config = config
    self.response_timeout = response_timeout

run async 

run(audio: Audio) -> AgentResponse

Send audio to a Live session and collect function calls.

Source code in src/russo/adapters/gemini.py 
159
160
161
162
163
164
165
166
167
168
169
async def run(self, audio: Audio) -> AgentResponse:
    """Send audio to a Live session and collect function calls."""
    if self.session is not None:
        return await self._run_on(self.session, audio)

    from google.genai import types

    assert self.client is not None
    config = self._build_config(types)
    async with self.client.aio.live.connect(model=self.model, config=config) as session:
        return await self._run_on(session, audio)