Google Chat Bot Library (gchatbot)

A modern Python library for building Google Chat bots, leveraging FastAPI for high performance and native asynchronous support.

Overview

This library provides a robust base class (GChatBot) that handles the complexities of the Google Chat API, allowing you to focus on your bot's logic.

• Serverless-Safe Architecture: Designed from the ground up to work reliably in serverless environments like Google Cloud Run and AWS Lambda using FastAPI's BackgroundTasks.
• Hybrid Sync/Async Support: Automatically detects whether your processing methods are sync or async and handles them correctly.
• Progressive Responses: Provide immediate feedback to the user for long-running tasks and then update the message with the final result.
• Modular Architecture: Decoupled components like EventParser, AsyncProcessor, and ResponseFactory allow for advanced customization.
• Simplified Event Parsing: Automatically converts the various Google Chat event payloads into a clean, predictable ExtractedEventData structure.
• Fully Typed: Complete type hints for a superior developer experience and robust code.

How It Works: The Serverless-Safe Model

The library implements a processing model that is both efficient and safe for serverless environments, preventing premature termination of background jobs.

1. Request Received: The FastAPI endpoint receives an event from the Google Chat API.
2. Delegate to Handler: The request is passed to the GChatBot.handleRequest method, along with FastAPI's backgroundTasks object.
3. Parse and Route: The event payload is parsed into a clean ExtractedEventData object. The library checks if the target handler (_processMessage or _processSlashCommand) is async or sync.
4. Execution:
   • async handlers are awaited directly.
   • sync handlers are run in a separate thread (asyncio.to_thread) to avoid blocking the event loop.
5. Response Handling:
   • Simple Response: If the handler returns a string, it is sent back immediately as a synchronous JSON response.
   • Progressive Response: If the handler returns a tuple (quick_message, detailed_coroutine): a. The quick_message is sent back immediately as a synchronous JSON response. b. The detailed_coroutine is added to FastAPI's backgroundTasks via background_tasks.add_task. FastAPI guarantees the execution of this task, even after the response has been sent.

This model ensures that long-running background jobs are not terminated prematurely by the serverless environment's lifecycle.

Flow Diagram

graph TD
    subgraph "Google Chat"
        A[Webhook Request]
    end

    subgraph "FastAPI Application"
        B(Route Endpoint)
        C{handleRequest}
    end

    subgraph "GChatBot - Phase 1: Analysis & Routing"
        D[Parser: extractEventData]
        E{Is handler async or sync?}
        F[Run sync handler in thread]
        G[Run async handler directly]
        H(Result: ResponseType)
    end
    
    subgraph "GChatBot - Phase 2: Response Delivery"
        I{Is result a<br/>Progressive Response?}
        J[Factory: format simple response]
        K[Return HTTP 200 OK with JSON]
        L[Factory: format quick response]
        M[Return HTTP 200 OK with JSON]
        N[<b>background_tasks.add_task</b>]
    end

    subgraph "Background Process (Guaranteed by FastAPI)"
        O(Processor: runAndAndUpdate)
        P[Create asyncio.Task from coroutine]
        Q(Processor: handleAsyncResponse)
        R[API Call: Create 'Processing' msg]
        S[Await the Task's completion]
        T[API Call: Update msg with final result]
    end

    A --> B --> C
    C --> D
    D --> E
    E -- Sync --> F
    E -- Async --> G
    F --> H
    G --> H
    H --> I

    I -- No --> J
    J --> K

    I -- Yes --> L
    L --> M
    I -- Yes --> N

    N -.-> O
    O --> P
    P --> Q
    Q --> R
    R --> S
    S --> T
    
    style K fill:#d4edda,stroke:#155724
    style M fill:#d4edda,stroke:#155724
    style N fill:#ffeeba,stroke:#856404
    style T fill:#d4edda,stroke:#155724

Installation

# Install the library with FastAPI dependencies (Recommended)
pip install "gchatbot[fastapi]"

Recommended Usage: FastAPI Example

# example.py
import os
import asyncio
from typing import Any, Dict
from fastapi import FastAPI, Request, BackgroundTasks
from gchatbot import GChatBot, ExtractedEventData, EventPayload, ResponseType

# Ensure you have a 'service.json' file or set the environment variable.
SERVICE_ACCOUNT_FILE: str = os.environ.get("SERVICE_ACCOUNT_FILE", "service.json")

class ExampleBot(GChatBot):
    """
    An example bot demonstrating synchronous and asynchronous methods.
    
    This example shows how you can mix sync and async methods in the same
    class, and the library will automatically detect and handle each appropriately.
    It also demonstrates serverless-safe progressive responses.
    """
    def __init__(self) -> None:
        super().__init__(
            botName="Hybrid Example Bot",
            serviceAccountFile=SERVICE_ACCOUNT_FILE,
        )

    async def _processSlashCommand(self, command: str, arguments: str, extractedData: ExtractedEventData, eventData: EventPayload) -> ResponseType:
        """
        ASYNCHRONOUS METHOD - Processes slash commands using async/await.
        This method is async, so it can use `await` for non-blocking operations.
        """
        user: str = extractedData.get('userDisplayName', 'User')
        userChatId: str = extractedData.get('userGoogleChatId', 'Unknown ID')
        
        if command == "long_task":
            # Demonstrates a serverless-safe progressive response.
            quickResponse = f"✅ Okay, {user}. I've started your long task. I will update this message when it's done."
            
            async def detailedResponse() -> str:
                await asyncio.sleep(8)  # Simulates a long-running API call or data processing.
                return f"🎉 Task complete for {user}! The results are ready."
            
            # Returning a tuple triggers the progressive response mechanism.
            return (quickResponse, detailedResponse())
        
        elif command == "info":
            # Demonstrates accessing the extracted event data.
            userEmail: str = extractedData.get('userEmail', 'Unknown Email')
            spaceName: str = extractedData.get('spaceName', 'Unknown Space')
            isDM: bool = extractedData.get('isDirectMessageEvent', False)
            
            return f"""ℹ️ *User and Context Info:*

*User:*
• Name: {user}
• Email: {userEmail}
• Google Chat ID: {userChatId}

*Space:*
• Name: {spaceName}
• Type: {'Direct Message (DM)' if isDM else 'Room/Group'}"""
        
        else:
            await asyncio.sleep(0.5)
            return f"✅ Unknown ASYNC command `/{command}` executed for {user}."

    def _processMessage(self, text: str, extractedData: ExtractedEventData, eventData: EventPayload) -> ResponseType:
        """
        SYNCHRONOUS METHOD - Processes regular messages using standard blocking calls.
        """
        user: str = extractedData.get('userDisplayName', 'User')
        
        if "help" in text.lower():
            return f"Hi {user}. I'm a demo bot. Try the slash commands `/long_task` or `/info`."
        
        else:
            return f"💬 SYNC message processed for {user}: '{text}'"

# --- FastAPI App Setup ---
app: FastAPI = FastAPI(title="gchatbot Hybrid Example")
bot: ExampleBot = ExampleBot()

@app.post("/webhook")
async def handleEvent(request: Request, backgroundTasks: BackgroundTasks) -> Any:
    """
    The main entry point for all Google Chat events.
    This function receives the `backgroundTasks` object from FastAPI
    and passes it to the GChatBot library.
    """
    return await bot.handleRequest(request, backgroundTasks)

@app.get("/")
def home() -> Dict[str, Any]:
    """
    A simple health-check endpoint.
    """
    return {
        "status": "active", 
        "bot_name": bot.botName, 
        "info": "This bot demonstrates both async and sync command handling.",
        "commands": [
            "/long_task - A slow command that uses a progressive response.",
            "/info - A quick command that shows user and space details."
        ]
    }

# To run locally: uvicorn example:app --reload --port 8080

Changelog

Version 0.3.1 - Code Quality & Documentation Updates

This minor version includes code quality improvements and documentation updates.

🔧 Improvements

• Code Quality: Removed deprecated syncTimeout parameter that was no longer used in the new BackgroundTasks architecture.
• Type Safety: Improved type annotations throughout the codebase for better development experience.
• Documentation: Updated README example to match the current example.py with more comprehensive demonstrations.
• Language Consistency: Continued standardization to English across the codebase.

Version 0.3.0 - Serverless Architecture Refactor

This version introduces a major architectural refactor to ensure robust performance in serverless environments and establishes new code quality standards.

💥 Breaking Changes

• Code Standard: camelCase: All method parameters and TypedDict keys have been standardized to camelCase (e.g., event_data is now eventData, background_tasks is backgroundTasks). You must update your method signatures when upgrading.
• Code Standard: English Language: All docstrings, comments, log messages, and default user-facing strings have been translated to English.

✨ New Features & Enhancements

• Serverless-Safe Architecture: The core handleRequest logic has been rebuilt to use FastAPI's BackgroundTasks. This guarantees that long-running processes (Progressive Responses) are not terminated prematurely in serverless environments like Google Cloud Run or AWS Lambda.
• Timeout Logic Removed: The old syncTimeout logic has been removed in favor of the more robust BackgroundTasks model. The bot now responds immediately with a quick message and schedules the long task, which is a more reliable pattern.
• Codebase Standardization: The entire library codebase now follows stricter standards for docstrings, typing, and naming conventions.

(Older changelog entries below)

License

This project is licensed under the MIT License. See the LICENSE file for details.