tomymaritano-chat-fastapi 0.1.0

FastAPI backend SDK for conversational AI chatbots with plugin architecture

tomymaritano-chat-fastapi

FastAPI backend SDK for conversational AI chatbots with plugin architecture

Enterprise-grade chatbot framework built with Clean Architecture, OpenAI integration, MongoDB storage, and extensible plugin system.

---

🚀 Installation

pip install tomymaritano-chat-fastapi

Optional dependencies:

# With Celery support
pip install tomymaritano-chat-fastapi[celery]

# With all extras
pip install tomymaritano-chat-fastapi[all]

---

📖 Quick Start

Basic Setup (5 minutes)

from fastapi import FastAPI
from tomymaritano_chat_fastapi import create_chat_router

app = FastAPI()

# Create chat router with minimal config
chat_router = create_chat_router(
    openai_api_key="sk-...",
    mongodb_url="mongodb://localhost:27017/chatbot_db"
)

# Mount router
app.include_router(chat_router, prefix="/api/chat")

# Run: uvicorn main:app --reload
# Test: POST http://localhost:8000/api/chat

That's it! Your chatbot API is ready.

---

🎯 Features

• ✅ Plugin System: Inject domain logic via ABCs (CommandHandler, ParserPlugin, ContextBuilder)
• ✅ Clean Architecture: Domain, Application, Infrastructure, API layers
• ✅ OpenAI Integration: Server-side AI conversations with context injection
• ✅ MongoDB Storage: Async Beanie ODM for conversations and quotas
• ✅ Auth & Quotas: PyJWT authentication + rate limiting
• ✅ Type Safe: Pydantic validation everywhere
• ✅ Production Ready: Structured logging, error handling, CORS
• ✅ Celery Support: Background jobs (optional)

---

🔌 Plugin System

Example: Custom Command Handler

from tomymaritano_chat_fastapi import CommandHandlerBase

class DolarCommandHandler(CommandHandlerBase):
    """Handle /dolar command - fetch exchange rates."""

    def can_handle(self, command: str) -> bool:
        return command in ['dolar', 'blue', 'oficial', 'mep']

    async def execute(self, command: str, args: list[str], context: dict) -> dict:
        # Fetch from your data source
        rate = await get_dolar_rate(command)

        return {
            "text": f"💵 **Dólar {command.title()}**: ${rate:,.2f}",
            "message_type": "command",
            "suggestions": [
                {"id": "1", "label": "Ver MEP", "command": "/mep"},
                {"id": "2", "label": "Ver CCL", "command": "/ccl"},
            ]
        }

# Register plugin
chat_router = create_chat_router(
    command_handlers=[DolarCommandHandler()],
    openai_api_key="sk-...",
    mongodb_url="mongodb://localhost:27017"
)

Example: Alert Parser Plugin

from tomymaritano_chat_fastapi import ParserPluginBase

class AlertParserPlugin(ParserPluginBase):
    """Parse natural language alerts like 'avisame cuando el dólar llegue a 1600'."""

    priority = 100  # Check before other parsers

    def is_intent(self, message: str) -> bool:
        keywords = ['avisame', 'alerta', 'notificame', 'aviso']
        return any(kw in message.lower() for kw in keywords)

    async def parse(self, message: str) -> dict:
        # Extract indicator, condition, target value using regex or NLP
        return {
            "intent": "create_alert",
            "indicator": "dolar_blue",
            "condition": "gte",  # >=
            "target": 1600
        }

# Register plugin
chat_router = create_chat_router(
    parsers=[AlertParserPlugin()],
    openai_api_key="sk-...",
    mongodb_url="mongodb://localhost:27017"
)

Example: Context Builder Plugin

from tomymaritano_chat_fastapi import ContextBuilderBase

class MarketContextBuilder(ContextBuilderBase):
    """Inject live market data into AI conversations."""

    async def build_context(self, user_id: str, metadata: dict) -> dict:
        # Fetch latest market data
        return {
            "dolar_blue": 1600,
            "dolar_oficial": 900,
            "embi": 1450,
            "inflation_monthly": 25.5
        }

    def format_for_prompt(self, context: dict) -> str:
        return f"""
        Datos actuales del mercado argentino:
        - Dólar Blue: ${context['dolar_blue']}
        - Dólar Oficial: ${context['dolar_oficial']}
        - Riesgo País: {context['embi']} puntos
        - Inflación mensual: {context['inflation_monthly']}%
        """

# Register plugin
chat_router = create_chat_router(
    context_builders=[MarketContextBuilder()],
    openai_api_key="sk-...",
    mongodb_url="mongodb://localhost:27017"
)

---

🛠️ Advanced Configuration

from tomymaritano_chat_fastapi import create_chat_router

chat_router = create_chat_router(
    # AI Configuration
    openai_api_key="sk-...",
    openai_model="gpt-4o-mini",  # or "gpt-4"
    openai_max_tokens=500,
    openai_temperature=0.7,

    # Database
    mongodb_url="mongodb://localhost:27017/chatbot_db",

    # Auth (optional)
    jwt_secret="your-secret-key",
    jwt_algorithm="HS256",
    require_auth=True,  # Require JWT auth for all requests

    # Plugins
    command_handlers=[...],
    parsers=[...],
    context_builders=[...],

    # Rate Limiting (optional)
    enable_rate_limiting=True,
    rate_limit_tier_limits={
        "free": 10,
        "basic": 50,
        "pro": -1  # unlimited
    },

    # CORS (optional)
    cors_origins=["http://localhost:3000"],

    # Celery (optional)
    celery_broker_url="redis://localhost:6379/0",
)

---

📡 API Contract

Endpoint: POST /api/chat

Request:

{
  "message": "¿Cuál es el precio del dólar blue?",
  "context": {
    "source": "web",
    "user_location": "AR"
  },
  "conversation_history": [
    {
      "id": "1",
      "role": "user",
      "content": "Hola",
      "timestamp": 1704067200000
    },
    {
      "id": "2",
      "role": "assistant",
      "content": "¡Hola! ¿En qué puedo ayudarte?",
      "timestamp": 1704067201000
    }
  ]
}

Response:

{
  "text": "💵 **Dólar Blue**: $1,600\n\n*Última actualización: hace 5 minutos*",
  "message_type": "command",
  "suggestions": [
    {
      "id": "1",
      "label": "Ver dólar oficial",
      "command": "/oficial",
      "icon": "💵"
    }
  ],
  "quota_remaining": 95
}

---

🗄️ Database Models

ConversationDocument

from tomymaritano_chat_fastapi import ConversationDocument

# Create conversation
conversation = ConversationDocument(
    user_id="user123",
    messages=[...],
    metadata={"source": "web"}
)
await conversation.save()

# Find user's conversations
conversations = await ConversationDocument.find(
    ConversationDocument.user_id == "user123"
).sort(-ConversationDocument.updated_at).to_list()

UserQuotaDocument

from tomymaritano_chat_fastapi import UserQuotaDocument

# Get user quota
quota = await UserQuotaDocument.find_one(UserQuotaDocument.user_id == "user123")

# Check remaining
if quota.has_quota_remaining():
    quota.increment_usage()
    await quota.save()

---

🔒 Authentication

JWT Authentication

from fastapi import Depends
from tomymaritano_chat_fastapi.api.dependencies import get_current_user

@app.get("/protected")
async def protected_route(user = Depends(get_current_user)):
    return {"user_id": user.user_id}

---

🧪 Testing

# Install dev dependencies
poetry install --with dev

# Run tests
poetry run pytest

# With coverage
poetry run pytest --cov=tomymaritano_chat_fastapi

# Type checking
poetry run mypy tomymaritano_chat_fastapi

---

🌐 Frontend Integration

Use @tomymaritano/chat-ui for the React frontend:

import { ChatWidget } from '@tomymaritano/chat-ui';

<ChatWidget
  apiEndpoint="http://localhost:8000/api/chat"
  title="Financial Assistant"
/>

---

📚 Examples

See examples/ directory:

• fastapi-example - Backend only
• full-stack-example - Backend + Frontend

---

📄 License

MIT © Tomy Maritano

---

🐛 Issues

Report bugs at: https://github.com/tomymaritano/tomymaritano-chat/issues