◈ API / SDK/2026-04-07Advanced

Gemini API × Slack Bot: Complete Production Guide — Bolt SDK, Thread Context, and Cloud Run Deployment

A complete guide to building a production-grade AI Slack Bot using Gemini API and Slack Bolt SDK (Python). Covers thread context management, multimodal support, rate limit handling, and Cloud Run deployment.

Gemini API¹⁸¹ Slack Slack Bot Python⁵² Cloud Run⁶ Bolt SDK chatbot⁴ production¹²⁴

✦ Premium Article

Setup and context — Why Gemini × Slack Bot?

Slack is the daily communication hub for countless teams worldwide. By combining it with Gemini API's advanced reasoning and multimodal capabilities, you can build an AI assistant that dramatically boosts team productivity — right inside the tools your team already uses.

In this guide, you'll learn everything you need to build and ship a production-ready Slack bot:

Setting up Slack Bolt SDK (Python) and configuring your Slack App
Integrating Gemini API with text, image, and file support
Managing conversation context across Slack threads
Handling rate limits, including 429 errors with exponential backoff
Secure credential management with Secret Manager
Deploying to Cloud Run and monitoring in production

This guide is aimed at intermediate-to-advanced developers with Python fundamentals who want to leverage Gemini API at a professional level.

Prerequisites and Required Tools

Here are the tools and versions used in this guide:

Python 3.11+
Slack Bolt for Python (slack_bolt >= 1.21)
Google GenAI SDK (google-genai >= 1.10)
Google Cloud project (with Cloud Run and Secret Manager enabled)
Slack workspace admin access (or permission to create Bot Tokens)

Install all required packages:

pip install slack_bolt google-genai google-cloud-secret-manager \
    google-cloud-logging fastapi uvicorn python-dotenv

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Build a production-grade Slack AI Bot from scratch using Gemini API and Slack Bolt SDK (Python)

✦Master implementation patterns for thread context management, multimodal support, and rate limit handling

✦Learn secure production deployment step by step using Cloud Run and Secret Manager

Secure payment via Stripe · Cancel anytime

Step 1: Configuring Your Slack App

1-1. Creating a Slack App

Head to api.slack.com/apps, click "Create New App" → "From scratch", and give your app a name and workspace.

Add the following Bot Token OAuth Scopes:

app_mentions:read — Read messages where the bot is mentioned
channels:history — Read message history in public channels
groups:history — Read message history in private channels
im:history — Read DM history
chat:write — Send messages
files:read — Access shared files

1-2. Setting Up Event Subscriptions

Enable Event Subscriptions and subscribe to these Bot Events:

message.channels — Messages in public channels
message.groups — Messages in private channels
message.im — Direct messages
app_mention — Mentions of your bot

You'll configure the Request URL after deploying to Cloud Run. During development, use a tunneling tool like ngrok.

1-3. Socket Mode for Local Development

For local development, Socket Mode lets you receive events without exposing a public URL. Enable Socket Mode and generate an App-Level Token with the connections:write scope.

Step 2: Project Structure

Here's a production-friendly directory layout:

gemini-slack-bot/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI entry point (production)
│   ├── bot.py           # Bolt app definition
│   ├── gemini_client.py # Gemini API wrapper
│   ├── context_store.py # Thread context management
│   └── config.py        # Settings and Secret Manager loader
├── Dockerfile
├── requirements.txt
└── .env.example

Step 3: Implementing the Gemini API Client

3-1. Gemini Client Wrapper

# app/gemini_client.py
import asyncio
import time
import logging
from typing import Optional
from google import genai
from google.genai import types
 
logger = logging.getLogger(__name__)
 
class GeminiClient:
    """
    Gemini API wrapper with built-in rate limiting and retry logic.
    """
 
    def __init__(self, api_key: str, model: str = "gemini-2.5-flash"):
        self.client = genai.Client(api_key=api_key)
        self.model = model
 
    async def generate(
        self,
        contents: list,
        system_instruction: Optional[str] = None,
        max_retries: int = 3,
    ) -> str:
        """
        Generate a text response from a list of content objects.
        Handles 429 (rate limit) errors with exponential backoff.
        """
        config = types.GenerateContentConfig(
            system_instruction=system_instruction or self._default_system_instruction(),
            max_output_tokens=4096,
            temperature=0.7,
        )
 
        for attempt in range(max_retries):
            try:
                response = await asyncio.to_thread(
                    self.client.models.generate_content,
                    model=self.model,
                    contents=contents,
                    config=config,
                )
                return response.text or ""
 
            except Exception as e:
                error_str = str(e)
                if "429" in error_str or "RESOURCE_EXHAUSTED" in error_str:
                    wait = 2 ** attempt * 5  # 5s, 10s, 20s
                    logger.warning(
                        f"Rate limited (attempt {attempt+1}/{max_retries}). "
                        f"Retrying in {wait}s..."
                    )
                    await asyncio.sleep(wait)
                    if attempt == max_retries - 1:
                        return "⚠️ The service is currently busy. Please try again in a moment."
                else:
                    logger.error(f"Gemini API error: {e}")
                    raise
 
        return "⚠️ An error occurred while generating a response."
 
    def _default_system_instruction(self) -> str:
        return (
            "You are an AI assistant operating within Slack. "
            "Answer team members' questions accurately and helpfully. "
            "Use Slack's markdown formatting (*bold*, _italic_, ```code blocks```) "
            "to make your responses easy to read. "
            "Be concise and use bullet points or numbered lists when appropriate."
        )

3-2. Multimodal Support (Image and File Processing)

# Additional method for app/gemini_client.py
 
    async def generate_with_image(
        self,
        text: str,
        image_bytes: bytes,
        mime_type: str = "image/png",
        history: Optional[list] = None,
    ) -> str:
        """
        Generate a response for an image + text prompt.
        Used to analyze images shared in Slack.
        """
        image_part = types.Part.from_bytes(data=image_bytes, mime_type=mime_type)
        text_part = types.Part.from_text(text=text)
 
        contents = []
        if history:
            contents.extend(history)
 
        contents.append(types.Content(
            role="user",
            parts=[image_part, text_part],
        ))
 
        return await self.generate(contents=contents)

Step 4: Thread Context Management

Managing conversation history per Slack thread is the key to natural, coherent dialogue. For production, persist context to Redis or Firestore — but let's start with an in-memory implementation.

# app/context_store.py
import time
import logging
from collections import defaultdict
from typing import Optional
from google.genai import types
 
logger = logging.getLogger(__name__)
 
MAX_HISTORY_TURNS = 10   # 10 turns (user + bot = 1 turn)
CONTEXT_TTL_SEC = 3600   # Expire context after 1 hour
 
class ThreadContextStore:
    """
    Stores Gemini conversation history keyed by Slack thread_ts.
 
    In production, replace with a Redis or Firestore backend.
    """
 
    def __init__(self):
        self._store: dict = defaultdict(
            lambda: {"history": [], "last_accessed": time.time()}
        )
 
    def get_history(self, thread_ts: str) -> list:
        """Return conversation history for a thread. Returns [] if expired."""
        entry = self._store.get(thread_ts)
        if not entry:
            return []
 
        if time.time() - entry["last_accessed"] > CONTEXT_TTL_SEC:
            logger.info(f"Context TTL expired for thread {thread_ts}")
            del self._store[thread_ts]
            return []
 
        return entry["history"]
 
    def add_turn(self, thread_ts: str, user_text: str, bot_text: str):
        """Append a user + bot turn to the conversation history."""
        entry = self._store[thread_ts]
        entry["last_accessed"] = time.time()
 
        entry["history"].append(
            types.Content(role="user", parts=[types.Part.from_text(text=user_text)])
        )
        entry["history"].append(
            types.Content(role="model", parts=[types.Part.from_text(text=bot_text)])
        )
 
        # Trim oldest turns if over limit
        while len(entry["history"]) > MAX_HISTORY_TURNS * 2:
            entry["history"].pop(0)
            entry["history"].pop(0)
 
    def clear(self, thread_ts: str):
        """Reset the context for a given thread."""
        if thread_ts in self._store:
            del self._store[thread_ts]
            logger.info(f"Context cleared for thread {thread_ts}")

Step 5: Implementing the Bolt App

5-1. Core Bot Logic

# app/bot.py
import logging
import httpx
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from app.gemini_client import GeminiClient
from app.context_store import ThreadContextStore
from app.config import Settings
 
logger = logging.getLogger(__name__)
 
settings = Settings()
 
app = AsyncApp(
    token=settings.slack_bot_token,
    signing_secret=settings.slack_signing_secret,
)
 
gemini = GeminiClient(api_key=settings.gemini_api_key)
context_store = ThreadContextStore()
handler = AsyncSlackRequestHandler(app)
 
def _get_thread_ts(event: dict) -> str:
    """Return the root thread_ts, or ts if the message is not in a thread."""
    return event.get("thread_ts") or event.get("ts", "")
 
@app.event("app_mention")
async def handle_mention(event: dict, say, client):
    """
    Handle @mentions of the bot.
    Carries over thread conversation context when inside a thread.
    """
    thread_ts = _get_thread_ts(event)
    channel_id = event["channel"]
    user_message = _strip_mention(event.get("text", ""))
 
    if not user_message.strip():
        await say(text="Hello! How can I help you today?", thread_ts=thread_ts)
        return
 
    # Reset context on "clear" command
    if user_message.strip().lower() in ["clear", "reset"]:
        context_store.clear(thread_ts)
        await say(text="✅ Conversation history has been cleared.", thread_ts=thread_ts)
        return
 
    files = event.get("files", [])
    image_file = next(
        (f for f in files if f.get("mimetype", "").startswith("image/")),
        None,
    )
 
    # Send a "thinking" indicator
    await client.chat_postMessage(
        channel=channel_id,
        thread_ts=thread_ts,
        text="⏳ Thinking...",
    )
 
    try:
        history = context_store.get_history(thread_ts)
 
        if image_file:
            image_bytes = await _download_slack_file(
                image_file["url_private"],
                settings.slack_bot_token,
            )
            response_text = await gemini.generate_with_image(
                text=user_message or "Please describe this image.",
                image_bytes=image_bytes,
                mime_type=image_file["mimetype"],
                history=history,
            )
        else:
            from google.genai import types
            contents = list(history) + [
                types.Content(
                    role="user",
                    parts=[types.Part.from_text(text=user_message)],
                )
            ]
            response_text = await gemini.generate(contents=contents)
 
        context_store.add_turn(thread_ts, user_message, response_text)
        await say(text=response_text, thread_ts=thread_ts)
 
    except Exception as e:
        logger.error(f"Error processing message: {e}", exc_info=True)
        await say(
            text="⚠️ An error occurred. Please try again in a moment.",
            thread_ts=thread_ts,
        )
 
@app.event("message")
async def handle_dm(event: dict, say):
    """
    Handle Direct Messages.
    In DMs, users can talk to the bot without @mentioning it.
    """
    if event.get("bot_id") or event.get("subtype"):
        return
 
    thread_ts = _get_thread_ts(event)
    user_message = event.get("text", "")
 
    if not user_message.strip():
        return
 
    try:
        from google.genai import types
        history = context_store.get_history(thread_ts)
        contents = list(history) + [
            types.Content(role="user", parts=[types.Part.from_text(text=user_message)])
        ]
        response_text = await gemini.generate(contents=contents)
        context_store.add_turn(thread_ts, user_message, response_text)
        await say(text=response_text, thread_ts=thread_ts)
 
    except Exception as e:
        logger.error(f"DM error: {e}", exc_info=True)
        await say(text="⚠️ An error occurred.", thread_ts=thread_ts)
 
def _strip_mention(text: str) -> str:
    """Remove <@BOTID> mentions from the text."""
    import re
    return re.sub(r"<@[A-Z0-9]+>", "", text).strip()
 
async def _download_slack_file(url: str, token: str) -> bytes:
    """Download a private Slack file and return its bytes."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            url,
            headers={"Authorization": f"Bearer {token}"},
            follow_redirects=True,
        )
        response.raise_for_status()
        return response.content

5-2. Configuration with Secret Manager

# app/config.py
import os
from functools import lru_cache
from pydantic_settings import BaseSettings
 
class Settings(BaseSettings):
    """
    Load settings from environment variables or .env file.
    In production, secrets are pulled from Secret Manager.
    """
    slack_bot_token: str = ""
    slack_signing_secret: str = ""
    slack_app_token: str = ""  # Socket Mode only (dev)
    gemini_api_key: str = ""
    environment: str = "development"
    gcp_project_id: str = ""
 
    class Config:
        env_file = ".env"
 
    def load_from_secret_manager(self):
        """Pull secrets from GCP Secret Manager in production."""
        if self.environment != "production":
            return
 
        from google.cloud import secretmanager
        client = secretmanager.SecretManagerServiceClient()
 
        def _get_secret(name: str) -> str:
            path = f"projects/{self.gcp_project_id}/secrets/{name}/versions/latest"
            response = client.access_secret_version(request={"name": path})
            return response.payload.data.decode("UTF-8")
 
        self.slack_bot_token = _get_secret("slack-bot-token")
        self.slack_signing_secret = _get_secret("slack-signing-secret")
        self.gemini_api_key = _get_secret("gemini-api-key")
 
@lru_cache()
def get_settings() -> Settings:
    s = Settings()
    s.load_from_secret_manager()
    return s

5-3. FastAPI Entry Point (Production)

# app/main.py
import logging
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.bot import handler
 
logging.basicConfig(level=logging.INFO)
api = FastAPI(title="Gemini Slack Bot")
 
@api.post("/slack/events")
async def slack_events(req: Request):
    """Slack Events API endpoint."""
    return await handler.handle(req)
 
@api.get("/health")
async def health():
    """Health check endpoint for Cloud Run."""
    return JSONResponse({"status": "ok"})

Step 6: Adding Slash Commands

Slash commands give your team an intuitive, discoverable interface for common bot interactions.

# Additional handlers for app/bot.py
 
@app.command("/gemini-help")
async def help_command(ack, respond):
    """Display usage instructions."""
    await ack()
    await respond(
        text=(
            "*🤖 Gemini Bot Usage*\n\n"
            "• `@GeminiBot [question]` — Ask anything\n"
            "• `@GeminiBot [question] + image` — Analyze an image\n"
            "• `@GeminiBot clear` — Reset conversation history\n"
            "• `/gemini-help` — Show this help message\n\n"
            "Mention the bot inside a thread to carry on the conversation with context."
        )
    )
 
@app.command("/gemini-summarize")
async def summarize_command(ack, respond, command):
    """
    Summarize provided text.
    Usage: /gemini-summarize [long text to summarize]
    """
    await ack()
    text = command.get("text", "").strip()
 
    if not text:
        await respond(text="Usage: `/gemini-summarize [text to summarize]`")
        return
 
    from google.genai import types
    from app.bot import gemini
 
    contents = [types.Content(
        role="user",
        parts=[types.Part.from_text(
            text=f"Please summarize the following text in 3–5 concise sentences:\n\n{text}"
        )],
    )]
    summary = await gemini.generate(contents=contents)
    await respond(text=f"*📝 Summary:*\n{summary}")

Step 7: Deploying to Cloud Run

7-1. Dockerfile

FROM python:3.11-slim
 
WORKDIR /app
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY app/ ./app/
 
ENV PORT=8080
EXPOSE 8080
 
CMD ["uvicorn", "app.main:api", "--host", "0.0.0.0", "--port", "8080", "--workers", "2"]

7-2. Registering Secrets in Secret Manager

# Register secrets (one-time setup)
gcloud secrets create slack-bot-token --replication-policy="automatic"
echo -n "xoxb-your-slack-bot-token" | \
  gcloud secrets versions add slack-bot-token --data-file=-
 
gcloud secrets create slack-signing-secret --replication-policy="automatic"
echo -n "your-signing-secret" | \
  gcloud secrets versions add slack-signing-secret --data-file=-
 
gcloud secrets create gemini-api-key --replication-policy="automatic"
echo -n "YOUR_GEMINI_API_KEY" | \
  gcloud secrets versions add gemini-api-key --data-file=-

7-3. Deploying to Cloud Run

PROJECT_ID="your-gcp-project-id"
REGION="us-central1"
SERVICE_NAME="gemini-slack-bot"
 
# Build and push the Docker image
gcloud builds submit --tag "gcr.io/${PROJECT_ID}/${SERVICE_NAME}"
 
# Deploy to Cloud Run
gcloud run deploy "${SERVICE_NAME}" \
  --image "gcr.io/${PROJECT_ID}/${SERVICE_NAME}" \
  --platform managed \
  --region "${REGION}" \
  --allow-unauthenticated \
  --memory 512Mi \
  --concurrency 80 \
  --timeout 60s \
  --set-env-vars "ENVIRONMENT=production,GCP_PROJECT_ID=${PROJECT_ID}" \
  --service-account "gemini-slack-bot@${PROJECT_ID}.iam.gserviceaccount.com"
 
# Grant Secret Manager access to the service account
gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member "serviceAccount:gemini-slack-bot@${PROJECT_ID}.iam.gserviceaccount.com" \
  --role "roles/secretmanager.secretAccessor"

After deployment, copy the Cloud Run URL and paste it into Slack's Event Subscriptions as the Request URL (e.g., https://your-service.run.app/slack/events).

Step 8: Production Best Practices

8-1. Integrating Cloud Logging

# app/main.py — add for production
import os
 
if os.getenv("ENVIRONMENT") == "production":
    import google.cloud.logging
    client = google.cloud.logging.Client()
    client.setup_logging()

8-2. Rate Limit Strategy

With multiple team members potentially hitting the bot simultaneously, a queuing strategy is essential. Here's a simple token bucket limiter:

# app/rate_limiter.py
import asyncio
import time
 
class TokenBucketLimiter:
    """
    Token bucket rate limiter. Adjust rpm to match your Gemini API plan.
    """
    def __init__(self, rpm: int = 60):
        self.rpm = rpm
        self.tokens = rpm
        self.last_refill = time.time()
        self._lock = asyncio.Lock()
 
    async def acquire(self):
        async with self._lock:
            now = time.time()
            elapsed = now - self.last_refill
            refill = elapsed * (self.rpm / 60.0)
            self.tokens = min(self.rpm, self.tokens + refill)
            self.last_refill = now
 
            if self.tokens >= 1:
                self.tokens -= 1
                return True
 
            # Not enough tokens — wait for refill
            wait_time = (1 - self.tokens) / (self.rpm / 60.0)
            await asyncio.sleep(wait_time)
            self.tokens = 0
            return True

8-3. Error Handling and Observability

Set up Cloud Monitoring alerts for error rates and latency spikes
Use Cloud Logging structured logs for all Gemini API calls and Slack events
Refer to the Gemini API Error Troubleshooting Guide for handling common API errors

For cost optimization, combining this bot with Gemini API × Redis Semantic Caching can dramatically reduce API costs by caching responses to frequently repeated questions.

Why Bring Gemini AI into Slack?

Slack is where your team already communicates. By embedding a Gemini-powered AI assistant directly into Slack, you eliminate context-switching and give everyone instant access to an intelligent helper — right where they work.

Whether it's answering technical questions, summarizing long documents, assisting with code reviews, or pulling data from external services, a Gemini Slack bot can dramatically boost your team's productivity.

In this guide, you'll build a fully functional Slack bot that can:

Respond to mentions with AI-generated answers
Maintain conversational context across threaded replies
Call external tools and APIs via Gemini's Function Calling
Handle errors and rate limits gracefully

What You'll Learn

How to create and configure a Slack App with the right permissions
Integrating the Gemini API with Slack Bolt for Python
Implementing thread-aware multi-turn conversations
Extending bot capabilities with Function Calling

Prerequisites

Basic experience with Python and API development
Admin access to a Slack workspace
Familiarity with REST APIs and environment variables

Setting Up Your Environment

What You'll Need

Google AI Studio API key — Get one from Google AI Studio
Slack workspace — With permission to install apps
Python 3.10+ — For modern type hint support
ngrok or Cloudflare Tunnel — For local webhook development (optional with Socket Mode)

For a detailed walkthrough on getting your Gemini API key, check out "[Gemini API Quickstart]((/articles/gemini-api/gemini-api-quickstart)."

Install Dependencies

pip install slack-bolt google-genai python-dotenv

Configure Environment Variables

Create a .env file in your project root:

# .env
SLACK_BOT_TOKEN=xoxb-your-bot-token
SLACK_APP_TOKEN=xapp-your-app-token
SLACK_SIGNING_SECRET=your-signing-secret
GEMINI_API_KEY=your-gemini-api-key

Creating and Configuring the Slack App

Step 1: Create a Slack App

Head to the Slack API dashboard and click "Create New App." Choose "From scratch," then enter a name and select your workspace.

Step 2: Set Bot Token Scopes

Under "OAuth & Permissions," add the following Bot Token Scopes:

app_mentions:read    — Read messages that mention the bot
chat:write           — Send messages
channels:history     — Read channel history (for thread context)
groups:history       — Read private channel history
im:history           — Read direct message history

Step 3: Enable Socket Mode

Socket Mode lets your bot receive events without a public URL — perfect for development. Enable it under "Socket Mode" and save the generated xapp- App-Level Token.

Step 4: Subscribe to Events

Under "Event Subscriptions," enable events and add app_mention to Bot Events. This triggers your handler whenever someone mentions the bot.

Building the Basic Gemini Slack Bot

Here's a minimal but functional bot that responds to mentions using the Gemini API:

import os
from dotenv import load_dotenv
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from google import genai
 
load_dotenv()
 
# Initialize Slack app
app = App(token=os.environ["SLACK_BOT_TOKEN"])
 
# Initialize Gemini client
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
SYSTEM_INSTRUCTION = """You are a helpful AI assistant for a development team.
Answer questions accurately and concisely.
Include code examples when answering programming questions.
Match the language of the user's message."""
 
@app.event("app_mention")
def handle_mention(event, say):
    """Respond to bot mentions"""
    user_message = event["text"]
    thread_ts = event.get("thread_ts", event["ts"])
 
    try:
        # Call the Gemini API
        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=user_message,
            config=genai.types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTION,
                max_output_tokens=2048,
                temperature=0.7,
            ),
        )
 
        # Reply in the thread
        say(text=response.text, thread_ts=thread_ts)
 
    except Exception as e:
        say(
            text=f"Sorry, an error occurred: {str(e)}",
            thread_ts=thread_ts,
        )
 
if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    print("⚡ Gemini Slack Bot is running")
    handler.start()

Save this as bot.py and run it with python bot.py. Mention your bot in any Slack channel, and it will respond with a Gemini-generated reply in the thread.

Adding Thread-Aware Multi-Turn Conversations

The basic implementation treats each message independently. To maintain context within a thread, fetch the conversation history and pass it to Gemini as a multi-turn conversation:

from google.genai import types
 
def get_thread_history(client_slack, channel, thread_ts):
    """Fetch thread history and convert to Gemini message format"""
    result = client_slack.conversations_replies(
        channel=channel, ts=thread_ts, limit=20
    )
 
    contents = []
    for msg in result["messages"]:
        # Determine if the message is from the bot or a user
        if msg.get("bot_id"):
            role = "model"
        else:
            role = "user"
 
        contents.append(
            types.Content(
                role=role,
                parts=[types.Part(text=msg["text"])],
            )
        )
 
    return contents
 
@app.event("app_mention")
def handle_mention_with_context(event, say, client):
    """Respond with full thread context"""
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])
 
    # Fetch thread history
    contents = get_thread_history(client, channel, thread_ts)
 
    try:
        gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
        response = gemini_client.models.generate_content(
            model="gemini-2.5-flash",
            contents=contents,
            config=genai.types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTION,
                max_output_tokens=2048,
                temperature=0.7,
            ),
        )
 
        say(text=response.text, thread_ts=thread_ts)
 
    except Exception as e:
        say(
            text=f"An error occurred: {str(e)}",
            thread_ts=thread_ts,
        )

Now when users ask follow-up questions in the same thread, the bot understands the full context of the conversation.

Extending with Function Calling

Gemini's [Function Calling]((/articles/gemini-advanced/function-calling-guide) feature lets your bot interact with external APIs and services. Here's an example that handles weather lookups and task creation:

from google.genai import types
 
# Define available tools
weather_tool = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="get_weather",
            description="Get current weather for a specified city",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "city": types.Schema(
                        type="STRING",
                        description="City name (e.g., Tokyo, New York)",
                    ),
                },
                required=["city"],
            ),
        ),
        types.FunctionDeclaration(
            name="create_task",
            description="Create a new task in the task management system",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "title": types.Schema(
                        type="STRING", description="Task title"
                    ),
                    "assignee": types.Schema(
                        type="STRING", description="Person assigned to the task"
                    ),
                    "priority": types.Schema(
                        type="STRING",
                        description="Task priority level",
                        enum=["high", "medium", "low"],
                    ),
                },
                required=["title"],
            ),
        ),
    ]
)
 
def execute_function(function_call):
    """Execute the function call and return results"""
    name = function_call.name
    args = function_call.args
 
    if name == "get_weather":
        # Replace with a real weather API call
        return {"city": args["city"], "temp": "72°F", "condition": "Sunny"}
    elif name == "create_task":
        # Replace with your actual task management API
        return {"status": "created", "title": args["title"]}
    else:
        return {"error": f"Unknown function: {name}"}
 
@app.event("app_mention")
def handle_with_tools(event, say, client):
    """Handle mentions with Function Calling support"""
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])
    user_message = event["text"]
 
    gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
    try:
        response = gemini_client.models.generate_content(
            model="gemini-2.5-flash",
            contents=user_message,
            config=genai.types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTION,
                tools=[weather_tool],
                temperature=0.7,
            ),
        )
 
        # Handle Function Call responses
        if response.candidates[0].content.parts[0].function_call:
            fc = response.candidates[0].content.parts[0].function_call
            result = execute_function(fc)
 
            # Pass results back to Gemini for a natural language response
            response = gemini_client.models.generate_content(
                model="gemini-2.5-flash",
                contents=[
                    types.Content(
                        role="user", parts=[types.Part(text=user_message)]
                    ),
                    response.candidates[0].content,
                    types.Content(
                        role="user",
                        parts=[
                            types.Part(
                                function_response=types.FunctionResponse(
                                    name=fc.name, response=result
                                )
                            )
                        ],
                    ),
                ],
                config=genai.types.GenerateContentConfig(
                    system_instruction=SYSTEM_INSTRUCTION,
                    tools=[weather_tool],
                ),
            )
 
        say(text=response.text, thread_ts=thread_ts)
 
    except Exception as e:
        say(text=f"An error occurred: {str(e)}", thread_ts=thread_ts)

For a deeper dive into Function Calling mechanics, see "[Gemini Function Calling — A Practical Guide]((/articles/gemini-advanced/function-calling-guide)."

Error Handling and Rate Limit Management

For a production-grade bot, robust error handling is essential:

import time
import logging
from functools import wraps
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
def retry_with_backoff(max_retries=3, base_delay=1.0):
    """Decorator for retrying API calls with exponential backoff"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    logger.warning(
                        f"Attempt {attempt + 1} failed: {e}. "
                        f"Retrying in {delay}s..."
                    )
                    time.sleep(delay)
        return wrapper
    return decorator
 
@retry_with_backoff(max_retries=3)
def call_gemini(contents, tools=None):
    """Gemini API call with automatic retry"""
    gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    config = genai.types.GenerateContentConfig(
        system_instruction=SYSTEM_INSTRUCTION,
        max_output_tokens=2048,
        temperature=0.7,
    )
    if tools:
        config.tools = tools
 
    return gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=contents,
        config=config,
    )

For a comprehensive guide to managing API quotas, see "[Gemini API Rate Limiting and Quota Management]((/articles/gemini-api/gemini-api-rate-limiting-quota-management)."

Handling Long Responses

Slack messages have a 40,000-character limit, but overly long responses hurt readability. Set max_output_tokens appropriately, or split long messages:

def split_message(text, max_length=3000):
    """Split long messages at natural break points"""
    if len(text) <= max_length:
        return [text]
 
    chunks = []
    while text:
        if len(text) <= max_length:
            chunks.append(text)
            break
        # Split at the nearest newline
        split_pos = text.rfind("\n", 0, max_length)
        if split_pos == -1:
            split_pos = max_length
        chunks.append(text[:split_pos])
        text = text[split_pos:].lstrip()
 
    return chunks

Deploying to Production

Once your bot is working locally, here's how to ship it.

Containerize with Docker

FROM python:3.12-slim
 
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
CMD ["python", "bot.py"]

Choosing a Hosting Platform

If you're using Socket Mode, you need a container that stays running. Google Compute Engine (GCE) or Cloud Run with "Always on CPU allocation" both work well.

For HTTP mode, switch Slack Bolt to an HTTP handler and point your Slack Event Subscriptions URL to your Cloud Run service URL. This is more cost-efficient since the container only runs when processing events.

Security Best Practices

Store API keys in environment variables or a secret manager — never hardcode them
Slack Bolt automatically validates requests using the Signing Secret
Grant only the minimum required OAuth scopes

Wrapping Up

In this guide, you built a team-facing AI assistant for Slack powered by the Gemini API. Starting from a simple mention responder, you progressively added thread-aware multi-turn conversations, Function Calling for external tool integration, and production-ready error handling.

Gemini 2.5 Flash's speed makes it ideal for real-time chat, and Function Calling opens the door to integrating virtually any API or service your team relies on. Start with the basics, gather feedback from your team, and iterate from there.

Summary

In this guide, we built a production-grade Slack AI bot powered by Gemini API. Here's a recap of the key takeaways:

Slack Bolt SDK with async handlers efficiently processes Slack events at scale
Thread-based context management (keyed on thread_ts) enables natural, multi-turn conversations
Exponential backoff on 429 errors ensures stable operation under load
Multimodal support lets users share images and get Gemini's analysis directly in Slack
Cloud Run + Secret Manager provides a secure, scalable, and cost-effective production environment

As next steps, consider extending the bot with Gemini Function Calling to enable real actions — like creating Jira tickets, opening GitHub Issues, or scheduling Google Calendar events — turning it from a Q&A assistant into a true team automation powerhouse.

Thank You for Reading

Gemini Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.