to Gemini API Function Calling: Tool Integration and Practical Usage

import google.generativeai as genai
import json
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
def get_current_weather(location: str, unit: str = "celsius") -> dict:
    """Calls a real weather API — using dummy data here"""
    return {
        "location": location,
        "temperature": 22 if unit == "celsius" else 71.6,
        "unit": unit,
        "condition": "Sunny",
        "humidity": 65
    }
 
def run_function_call_example():
    model = genai.GenerativeModel(
        model_name="gemini-2.0-flash",
        tools=[get_current_weather]  # Pass function directly — schema is auto-generated
    )
    
    chat = model.start_chat()
    response = chat.send_message("What's the weather in Tokyo right now?")
    
    # Process function calls until none remain
    while response.candidates[0].content.parts[0].function_call.name:
        fc = response.candidates[0].content.parts[0].function_call
        
        # Execute the tool
        if fc.name == "get_current_weather":
            result = get_current_weather(**fc.args)
        
        # Return results to the model
        response = chat.send_message(
            genai.protos.Content(
                parts=[genai.protos.Part(
                    function_response=genai.protos.FunctionResponse(
                        name=fc.name,
                        response={"result": json.dumps(result)}
                    )
                )]
            )
        )
    
    return response.text

def handle_parallel_function_calls(response, tools_map):
    """Generic handler for parallel tool invocations"""
    function_calls = [
        part.function_call 
        for part in response.candidates[0].content.parts
        if hasattr(part, 'function_call') and part.function_call.name
    ]
    
    if not function_calls:
        return None
    
    # Execute all tools (use asyncio or ThreadPool in production)
    results = []
    for fc in function_calls:
        if fc.name in tools_map:
            result = tools_map[fc.name](**fc.args)
            results.append(
                genai.protos.Part(
                    function_response=genai.protos.FunctionResponse(
                        name=fc.name,
                        response={"result": json.dumps(result)}
                    )
                )
            )
    
    return genai.protos.Content(parts=results)

from google.generativeai.types import ToolConfig, FunctionCallingConfig
 
# Mode 1: AUTO (default) — model decides on its own
auto_config = ToolConfig(
    function_calling_config=FunctionCallingConfig(mode="AUTO")
)
 
# Mode 2: ANY — must use at least one tool (forced)
forced_config = ToolConfig(
    function_calling_config=FunctionCallingConfig(
        mode="ANY",
        allowed_function_names=["get_current_weather", "search_database"]
    )
)
 
# Mode 3: NONE — no tool use (text response only)
no_tools_config = ToolConfig(
    function_calling_config=FunctionCallingConfig(mode="NONE")
)

class DataAnalysisAssistant:
    def __init__(self):
        self.tools = [
            self.query_database,
            self.calculate_statistics,
            self.create_visualization,
            self.export_report
        ]
        self.model = genai.GenerativeModel(
            model_name="gemini-2.0-pro",
            tools=self.tools,
            system_instruction="""
You are a data analysis expert.
Use tools as needed to answer the user's questions.
When multiple tools are required, use them in logical order.
            """
        )
        self.chat = self.model.start_chat()
    
    def analyze(self, user_query: str) -> str:
        response = self.chat.send_message(user_query)
        
        # Loop until no more tool calls
        while True:
            function_calls = self._extract_function_calls(response)
            if not function_calls:
                break
            
            results = self._execute_function_calls(function_calls)
            response = self.chat.send_message(results)
        
        return response.text

def safe_execute_tool(tool_name: str, args: dict, tools_map: dict) -> dict:
    """Tool execution wrapper with graceful error handling"""
    try:
        if tool_name not in tools_map:
            return {
                "error": f"Tool '{tool_name}' not found",
                "available_tools": list(tools_map.keys())
            }
        
        result = tools_map[tool_name](**args)
        return {"success": True, "result": result}
        
    except ValueError as e:
        return {
            "error": "Invalid argument value",
            "details": str(e),
            "suggestion": "Check argument types and acceptable ranges"
        }
    except TimeoutError:
        return {
            "error": "Tool execution timed out",
            "suggestion": "Try again later or consider a different approach"
        }
    except Exception as e:
        return {
            "error": "Unexpected error occurred",
            "details": str(e)
        }

import logging
from functools import wraps
 
def audit_tool_call(func):
    """Decorator that logs every tool invocation"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(
            f"Tool called: {func.__name__}, "
            f"args: {args}, kwargs: {kwargs}"
        )
        try:
            result = func(*args, **kwargs)
            logging.info(f"Tool {func.__name__} succeeded")
            return result
        except Exception as e:
            logging.error(f"Tool {func.__name__} failed: {e}")
            raise
    return wrapper
 
@audit_tool_call
def sensitive_database_query(query: str) -> list:
    # Database query implementation
    pass

customer_support_tools = [
    {
        "name": "get_order_status",
        "description": "Retrieves order status, shipping info, and estimated delivery date for a given order number",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "Order number (e.g., ORD-2026-12345)"}
            },
            "required": ["order_id"]
        }
    },
    {
        "name": "process_return_request",
        "description": "Processes a return or exchange request. Only use after getting explicit user confirmation.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string"},
                "reason": {"type": "string"},
                "type": {"type": "string", "enum": ["return", "exchange"]}
            },
            "required": ["order_id", "reason", "type"]
        }
    }
]

code_review_tools = [
    "analyze_code_complexity",           # Measure code complexity
    "check_security_vulnerabilities",    # Scan for security issues
    "suggest_refactoring",               # Propose refactoring opportunities
    "run_tests",                         # Execute test suite
    "check_dependency_updates"           # Check for dependency updates
]

import asyncio
 
async def execute_parallel_tools(function_calls: list, tools_map: dict) -> list:
    """Execute multiple tools in parallel using async"""
    async def execute_single(fc):
        tool = tools_map.get(fc.name)
        if not tool:
            return {"error": f"Unknown tool: {fc.name}"}
        
        if asyncio.iscoroutinefunction(tool):
            result = await tool(**fc.args)
        else:
            # Run synchronous functions in a thread pool
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(None, lambda: tool(**fc.args))
        
        return result
    
    return await asyncio.gather(*[execute_single(fc) for fc in function_calls])

import google.generativeai as genai
 
# Configure your API key
genai.configure(api_key="YOUR_API_KEY")
 
# Initialize the model
model = genai.GenerativeModel("gemini-2.5-pro")
 
# Receive streaming response
response = model.generate_content(
    "Explain how to build a web scraper in Python, step by step.",
    stream=True
)
 
# Output each chunk in real time
for chunk in response:
    if chunk.text:
        print(chunk.text, end="", flush=True)
 
print()  # Final newline
 
# Expected behavior:
# Text appears in real time (token by token)
# No waiting for the complete response to generate

import { GoogleGenerativeAI } from "@google/generative-ai";
 
const genAI = new GoogleGenerativeAI("YOUR_API_KEY");
const model = genAI.getGenerativeModel({ model: "gemini-2.5-pro" });
 
async function streamResponse(prompt: string) {
  const result = await model.generateContentStream(prompt);
  
  // Process chunks as they arrive
  for await (const chunk of result.stream) {
    const text = chunk.text();
    process.stdout.write(text);
  }
  
  // You can also get the complete final response
  const finalResponse = await result.response;
  console.log("\n\nTotal tokens:", finalResponse.usageMetadata?.totalTokenCount);
}
 
// Usage
streamResponse("What are the best practices for designing REST APIs in TypeScript?");
 
// Expected output:
// (Text streams in real time)
// Total tokens: 1234

import google.generativeai as genai
 
genai.configure(api_key="YOUR_API_KEY")
 
model = genai.GenerativeModel(
    model_name="gemini-2.5-pro",
    tools=[get_weather]
)
 
# Streaming + Function Calling
response = model.generate_content(
    "Help me plan a picnic for tomorrow. Check Tokyo's weather and create an appropriate packing list.",
    stream=True
)
 
for chunk in response:
    # Handle Function Call requests in stream
    for part in chunk.parts:
        if hasattr(part, 'function_call'):
            result = get_weather(city="Tokyo")
            print(f"\n[Weather data fetched: {result}]\n")
    
    # Handle text responses
    if chunk.text:
        print(chunk.text, end="", flush=True)