I'm working with Gemini Live API and I want to load previous conversation context to my current live session, allowing me to resume any session from the local save. So far I've come up with this code:
import asyncio
from pathlib import Path
import sys
import json
from google import genai
from google.genai import types
key_path = Path("./api-key.txt")
if not key_path.exists():
key_path.touch()
with open(key_path, "r") as f:
key = f.readline().strip()
if not key:
print(f"Paste your API key into {key_path.resolve()}")
sys.exit(1)
client = genai.Client(
api_key=key,
http_options={'api_version': 'v1beta'}
)
MODEL = "gemini-2.5-flash-native-audio-preview-12-2025"
get_weather_tool = {
"name": "getWeather",
"description": "gets the weather (temperature in Celsius) for a requested city",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"}
},
"required": ["city"]
}
}
tools_config = [types.Tool(function_declarations=[get_weather_tool])]
CONFIG = types.LiveConnectConfig(
response_modalities=["AUDIO"],
speech_config=types.SpeechConfig(
voice_config=types.VoiceConfig(
prebuilt_voice_config=types.PrebuiltVoiceConfig(
voice_name="Zephyr")
)
),
tools=tools_config,
)
async def run():
try:
async with client.aio.live.connect(model=MODEL, config=CONFIG) as session:
print("Connected! Loading context...")
func_call_id = "call_12345"
history_turns = [
# 1. User
types.Content(
role="user",
parts=[types.Part(
text="Hello, my name is Sasha. My hobby is skiing")]
),
# 2. Model
types.Content(
role="model",
parts=[types.Part(text="Hello Sasha! Nice to meet you.")]
),
# 3. User
types.Content(
role="user",
parts=[types.Part(text="What is the weather in Kyiv?")]
),
# 4. Function call
types.Content(
role="model",
parts=[types.Part(
function_call=types.FunctionCall(
name="getWeather",
args={"city": "Kyiv"},
id=func_call_id
)
)]
),
# 5. Tool Response
types.Content(
role="user",
parts=[types.Part(
function_response=types.FunctionResponse(
name="getWeather",
response={"result": "25"},
id=func_call_id
)
)]
),
# 6. Model
types.Content(
role="model",
parts=[types.Part(text="It's 25 degrees celsius in Kyiv.")]
),
]
await session.send_client_content(
turns=history_turns,
turn_complete=False
)
print("Context loaded.")
while True:
text = await asyncio.to_thread(input, "\nYou: ")
if text.lower() == "q":
break
await session.send_client_content(
turns=[types.Content(
role="user",
parts=[types.Part(text=text)]
)],
turn_complete=True
)
print("Gemini: ", end="")
async for response in session.receive():
if response.server_content:
if response.server_content.output_transcription:
print(
response.server_content.output_transcription.text, end="", flush=True)
if response.server_content.model_turn:
for part in response.server_content.model_turn.parts:
if part.text:
print(part.text, end="", flush=True)
if response.server_content and response.server_content.turn_complete:
break
print()
except Exception as e:
print("\nConnection failed/closed:", repr(e))
if __name__ == "__main__":
asyncio.run(run())
When I run it and type any prompt, I get this error:
Connected! Loading context...
Context loaded.
You: What was the weather like?
Connection failed/closed: ConnectionClosedError(Close(code=1007, reason='Request contains an invalid argument.'), Close(code=1007, reason='Request contains an invalid argument.'), True)
It fails at the line async for response in session.receive():,
But it's clearly related to this part:
# 4. Function call
types.Content(
role="model",
parts=[types.Part(
function_call=types.FunctionCall(
name="getWeather",
args={"city": "Kyiv"},
id=func_call_id
)
)]
),
# 5. Tool Response
types.Content(
role="user",
parts=[types.Part(
function_response=types.FunctionResponse(
name="getWeather",
response={"result": "25"},
id=func_call_id
)
)]
),
because if I remove it, the execution passes and the model can tell that my hobby is skiing for example.
I have also tried wrapping {"city": "Kyiv"} and {"result": "25"} into json.dumps(), but I get another error:
Connection failed/closed: 1 validation error for FunctionCall
args
Input should be a valid dictionary [type=dict_type, input_value='{"city": "Kyiv"}', input_type=str]
For further information visit https://errors.pydantic.dev/2.12/v/dict_type
Is there a way to pass tool calls into newly created conversation? Or what is the intended way of having sessions saved forever in the Gemini Live API context?
The issue is likely that Gemini Live API does not yet support loading history with structured function calls and responses—it's a limitation in the preview models. Remove those parts ( #4 and #5) from history_turns or convert them to text parts for context:
# 4. Simulate function call as text
types.Content(
role="model",
parts=[types.Part(text="Calling getWeather for Kyiv.")]
),
# 5. Simulate tool response as text
types.Content(
role="user",
parts=[types.Part(text="The result is 25 degrees Celsius.")]
),
This loads the context without error. For persistent sessions, summarize long histories to fit token limits (128k for native-audio models).
To check available models (names may differ by API key/account), use:
import google.generativeai as genai
genai.configure(api_key=key)
for m in genai.list_models():
print(m.name, m.display_name)
What models does list_models() show for you? Exact names like "models/gemini-1.5-flash" or "models/gemini-2.0-experimental" are needed for Live API.