Docs › Google Gemini

Google Gemini API Integration

Google Gemini is a family of multimodal LLMs from Google DeepMind, available via the google-generativeai Python package and @google/generative-ai TypeScript package. This guide covers tracing generate_content() calls, multi-turn chat sessions, function calling, token budget tracking, and debugging blocked responses.

Installation

Python

pip install keylightdigital-nexus google-generativeai

TypeScript

npm install @keylightdigital/nexus @google/generative-ai

Get your Nexus API key from Dashboard → API Keys and set it alongside your Gemini key:

export NEXUS_API_KEY="nxs_your_key_here"
export GEMINI_API_KEY="AIza..."

Basic generate_content() trace

Wrap each generate_content() call in a Nexus trace. Every request appears in the dashboard with its model, prompt preview, status, and latency:

Python

import os
import google.generativeai as genai
from nexus_sdk import NexusClient

genai.configure(api_key=os.environ["GEMINI_API_KEY"])
nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

model = genai.GenerativeModel("gemini-1.5-pro")

def ask(prompt: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "gemini-assistant",
        "name": f"gemini: {prompt[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {
            "model": "gemini-1.5-pro",
            "prompt_preview": prompt[:200],
        },
    })
    try:
        response = model.generate_content(prompt)
        finish_reason = response.candidates[0].finish_reason.name
        nexus.end_trace(trace["trace_id"], {
            "status": "success",
            "metadata": {
                "prompt_tokens": response.usage_metadata.prompt_token_count,
                "completion_tokens": response.usage_metadata.candidates_token_count,
                "finish_reason": finish_reason,
            },
        })
        return response.text
    except Exception as e:
        nexus.end_trace(trace["trace_id"], {
            "status": "error",
            "metadata": {"error": str(e)},
        })
        raise

TypeScript

import { GoogleGenerativeAI } from '@google/generative-ai';
import { NexusClient } from '@keylightdigital/nexus';

const genAI = new GoogleGenerativeAI(process.env.GEMINI_API_KEY!);
const nexus = new NexusClient({
  apiKey: process.env.NEXUS_API_KEY!,
  agentId: 'gemini-assistant',
});
const model = genAI.getGenerativeModel({ model: 'gemini-1.5-pro' });

async function ask(prompt: string): Promise<string> {
  const trace = await nexus.startTrace({
    agentId: 'gemini-assistant',
    input: prompt,
  });
  try {
    const result = await model.generateContent(prompt);
    const response = result.response;
    const meta = response.usageMetadata;
    await nexus.endTrace(trace.id, {
      output: response.text(),
      status: 'success',
      metadata: {
        promptTokens: meta?.promptTokenCount,
        completionTokens: meta?.candidatesTokenCount,
        finishReason: response.candidates?.[0]?.finishReason,
        model: 'gemini-1.5-pro',
      },
    });
    return response.text();
  } catch (err) {
    await nexus.endTrace(trace.id, { output: String(err), status: 'error' });
    throw err;
  }
}

Multi-turn chat sessions

Gemini’s ChatSession maintains conversation history automatically. Open one trace per session and add a span per turn so you can see the full conversation waterfall in Nexus:

Python

import os
import google.generativeai as genai
from nexus_sdk import NexusClient

genai.configure(api_key=os.environ["GEMINI_API_KEY"])
nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

model = genai.GenerativeModel("gemini-1.5-flash")

def run_chat(messages: list[str]) -> list[str]:
    trace = nexus.start_trace({
        "agent_id": "gemini-chat",
        "name": "chat-session",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {"model": "gemini-1.5-flash", "turns": len(messages)},
    })
    chat = model.start_chat(history=[])
    replies = []
    try:
        for i, message in enumerate(messages):
            span = nexus.start_span(trace["trace_id"], {
                "name": f"turn-{i + 1}",
                "type": "llm",
                "metadata": {
                    "turn": i + 1,
                    "user_message": message[:200],
                },
            })
            response = chat.send_message(message)
            nexus.end_span(span["id"], {
                "output": response.text[:500],
                "metadata": {
                    "prompt_tokens": response.usage_metadata.prompt_token_count,
                    "completion_tokens": response.usage_metadata.candidates_token_count,
                },
            })
            replies.append(response.text)
        nexus.end_trace(trace["trace_id"], {"status": "success"})
        return replies
    except Exception as e:
        nexus.end_trace(trace["trace_id"], {
            "status": "error",
            "metadata": {"error": str(e)},
        })
        raise

TypeScript

import { GoogleGenerativeAI } from '@google/generative-ai';
import { NexusClient } from '@keylightdigital/nexus';

const genAI = new GoogleGenerativeAI(process.env.GEMINI_API_KEY!);
const nexus = new NexusClient({ apiKey: process.env.NEXUS_API_KEY!, agentId: 'gemini-chat' });
const model = genAI.getGenerativeModel({ model: 'gemini-1.5-flash' });

async function runChat(messages: string[]): Promise<string[]> {
  const trace = await nexus.startTrace({
    agentId: 'gemini-chat',
    input: messages[0],
  });
  const chat = model.startChat({ history: [] });
  const replies: string[] = [];
  try {
    for (let i = 0; i < messages.length; i++) {
      const span = await nexus.startSpan(trace.id, {
        name: `turn-${i + 1}`,
        type: 'llm',
        metadata: { turn: i + 1, userMessage: messages[i] },
      });
      const result = await chat.sendMessage(messages[i]);
      const response = result.response;
      await nexus.endSpan(span.id, {
        output: response.text().slice(0, 500),
        metadata: {
          promptTokens: response.usageMetadata?.promptTokenCount,
          completionTokens: response.usageMetadata?.candidatesTokenCount,
        },
      });
      replies.push(response.text());
    }
    await nexus.endTrace(trace.id, { output: replies.at(-1) ?? '', status: 'success' });
    return replies;
  } catch (err) {
    await nexus.endTrace(trace.id, { output: String(err), status: 'error' });
    throw err;
  }
}

Function calling

Gemini supports function calling via tool_config and function declarations. Add a Nexus span for each function invocation to see tool inputs, outputs, and errors in the trace waterfall:

Python — function calling loop

import os
import google.generativeai as genai
from nexus_sdk import NexusClient

genai.configure(api_key=os.environ["GEMINI_API_KEY"])
nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

get_weather_decl = genai.protos.FunctionDeclaration(
    name="get_weather",
    description="Get current weather for a city",
    parameters=genai.protos.Schema(
        type=genai.protos.Type.OBJECT,
        properties={"city": genai.protos.Schema(type=genai.protos.Type.STRING)},
        required=["city"],
    ),
)

model = genai.GenerativeModel(
    "gemini-1.5-pro",
    tools=[genai.protos.Tool(function_declarations=[get_weather_decl])],
)

def fetch_weather(city: str) -> dict:
    return {"city": city, "temperature": "22C", "condition": "Sunny"}

def run_agent(task: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "gemini-function-agent",
        "name": f"agent: {task[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {"model": "gemini-1.5-pro", "task": task[:200]},
    })
    try:
        response = model.generate_content(task)
        # Agentic loop: handle function calls until the model stops
        while True:
            candidate = response.candidates[0]
            fn_parts = [p for p in candidate.content.parts if hasattr(p, "function_call")]
            if not fn_parts or candidate.finish_reason.name == "STOP":
                break
            fn = fn_parts[0].function_call
            fn_span = nexus.start_span(trace["trace_id"], {
                "name": f"tool:{fn.name}",
                "type": "tool",
                "metadata": {"function": fn.name, "args": dict(fn.args)},
            })
            result = fetch_weather(fn.args["city"]) if fn.name == "get_weather" else {"error": "unknown function"}
            nexus.end_span(fn_span["id"], {"output": result})
            # Feed function result back to Gemini
            response = model.generate_content([
                task,
                candidate.content,
                genai.protos.Content(parts=[
                    genai.protos.Part(
                        function_response=genai.protos.FunctionResponse(
                            name=fn.name,
                            response={"result": result},
                        )
                    )
                ]),
            ])
        nexus.end_trace(trace["trace_id"], {"status": "success"})
        return response.text
    except Exception as e:
        nexus.end_trace(trace["trace_id"], {"status": "error", "metadata": {"error": str(e)}})
        raise

TypeScript — function calling loop

import { GoogleGenerativeAI, SchemaType, type FunctionDeclaration } from '@google/generative-ai';
import { NexusClient } from '@keylightdigital/nexus';

const genAI = new GoogleGenerativeAI(process.env.GEMINI_API_KEY!);
const nexus = new NexusClient({ apiKey: process.env.NEXUS_API_KEY!, agentId: 'gemini-function-agent' });

const getWeatherDecl: FunctionDeclaration = {
  name: 'get_weather',
  description: 'Get current weather for a city',
  parameters: {
    type: SchemaType.OBJECT,
    properties: { city: { type: SchemaType.STRING, description: 'City name' } },
    required: ['city'],
  },
};

const model = genAI.getGenerativeModel({
  model: 'gemini-1.5-pro',
  tools: [{ functionDeclarations: [getWeatherDecl] }],
});

function fetchWeather(city: string) {
  return { city, temperature: '22C', condition: 'Sunny' };
}

async function runAgent(task: string): Promise<string> {
  const trace = await nexus.startTrace({ agentId: 'gemini-function-agent', input: task });
  try {
    let result = await model.generateContent(task);
    // Agentic loop
    while (true) {
      const candidate = result.response.candidates?.[0];
      const fnCall = candidate?.content.parts.find(p => p.functionCall)?.functionCall;
      if (!fnCall) break;

      const span = await nexus.startSpan(trace.id, {
        name: `tool:${fnCall.name}`,
        type: 'tool',
        metadata: { function: fnCall.name, args: fnCall.args },
      });
      const fnResult = fnCall.name === 'get_weather'
        ? fetchWeather((fnCall.args as { city: string }).city)
        : { error: 'unknown function' };
      await nexus.endSpan(span.id, { output: JSON.stringify(fnResult) });

      result = await model.generateContent([
        task,
        { role: 'model', parts: candidate!.content.parts },
        {
          role: 'user',
          parts: [{ functionResponse: { name: fnCall.name, response: fnResult } }],
        },
      ]);
    }
    await nexus.endTrace(trace.id, { output: result.response.text(), status: 'success' });
    return result.response.text();
  } catch (err) {
    await nexus.endTrace(trace.id, { output: String(err), status: 'error' });
    throw err;
  }
}

Token tracking

Every Gemini response includes usage_metadata with prompt and completion token counts. Log these on every span to build cost dashboards and catch runaway agents before they exhaust your quota:

# Python: full token metadata on every LLM span
span = nexus.start_span(trace["trace_id"], {
    "name": "gemini:generate",
    "type": "llm",
    "metadata": {"model": "gemini-1.5-pro", "prompt_preview": prompt[:200]},
})
response = model.generate_content(prompt)
nexus.end_span(span["id"], {
    "output": response.text[:500],
    "metadata": {
        "prompt_tokens": response.usage_metadata.prompt_token_count,
        "completion_tokens": response.usage_metadata.candidates_token_count,
        "total_tokens": response.usage_metadata.total_token_count,
        # Approximate cost: gemini-1.5-pro $1.25/1M input, $5.00/1M output
        "estimated_cost_usd": round(
            response.usage_metadata.prompt_token_count * 1.25e-6 +
            response.usage_metadata.candidates_token_count * 5.00e-6,
            8,
        ),
    },
})

Use Nexus trace metadata filters to alert when a session exceeds a token budget, or to compare token efficiency across model tiers: gemini-1.5-flash costs 15× less than gemini-1.5-pro for input tokens.

Debugging patterns

Response blocked: SAFETY finish reason

Gemini refuses requests that violate its safety policy. The candidate’s finish_reason will be SAFETY and calling response.text will raise a ValueError. Record the finish reason and per-category safety ratings as span metadata so you can identify which prompts trigger blocks:

# Python: detect and log blocked responses
span = nexus.start_span(trace["trace_id"], {"name": "gemini:generate", "type": "llm"})
response = model.generate_content(prompt)
candidate = response.candidates[0]
finish_reason = candidate.finish_reason.name  # e.g. "SAFETY", "RECITATION"

if finish_reason not in ("STOP", "MAX_TOKENS"):
    safety_ratings = [
        {"category": r.category.name, "probability": r.probability.name}
        for r in candidate.safety_ratings
    ]
    nexus.end_span(span["id"], {
        "status": "error",
        "metadata": {"finish_reason": finish_reason, "safety_ratings": safety_ratings},
    })
    nexus.end_trace(trace["trace_id"], {
        "status": "error",
        "metadata": {"blocked_reason": finish_reason},
    })
    raise ValueError(f"Gemini blocked response: {finish_reason}")

nexus.end_span(span["id"], {"output": response.text[:500]})

Response blocked: RECITATION finish reason

Gemini uses RECITATION when it would reproduce copyrighted content verbatim. If you see this in Nexus spans, rephrase your prompt to ask for a summary or paraphrase rather than quoting source material directly. Safety thresholds cannot override citation detection.

Function calling loop not terminating

If the agentic loop keeps generating function calls without reaching STOP, check the tool:* spans in Nexus to confirm that your function response was sent back with the exact name from the original function_call. A name mismatch causes the model to ignore the result and repeat the call.

Traces stuck in “running” state

A trace stays running if nexus.end_trace() is never called. Always wrap generate_content() and chat.send_message() in a try/except (Python) or try/catch (TypeScript) so the trace closes even when an exception propagates.

Ready to instrument your Gemini applications?

Start for free — no credit card required. See traces in under 5 minutes.

Start free →