Agente de SMS con IA

Un agente de SMS autónomo con IA. Los SMS entrantes llegan a través de la pasarela SMS TextBee, un webhook de FastAPI los entrega a un agente de IA local, que puede usar herramientas MCP cuando sea necesario, y la respuesta de la IA se envía de vuelta al mismo número a través de TextBee.

Estado: Pipeline completo conectado y endurecido para producción: webhook de TextBee -> SQLite -> worker de fondo asíncrono -> IA local (Ollama) -> herramientas MCP -> división de respuestas adaptada a SMS -> envío de TextBee -> estado de entrega.

Guía de producción (despliegue, HTTPS, seguridad, copias de seguridad, resolución de problemas): docs/PRODUCTION.md

Arquitectura

User SMS
  → TextBee SMS gateway
  → FastAPI webhook
  → Local AI agent
  → MCP tools when needed
  → AI generates response in the same language
  → TextBee
  → Response SMS to the same user

Requisitos

• Python 3.11+

• SQLite (incluido con Python)

Instalación

cd sms-ai-agent
python -m venv .venv

# Windows
.venv\Scripts\activate
# macOS / Linux
source .venv/bin/activate

pip install -r requirements.txt

Configuración

Copia el archivo de entorno de ejemplo y rellena tus valores:

cp .env.example .env

Ajustes de TextBee (https://textbee.dev):

Ajustes de Ollama:

Ajustes de MCP:

El archivo de base de datos SQLite se crea automáticamente en database/sms_agent.db en el primer arranque.

Ejecución

uvicorn app.main:app --reload

A continuación, comprueba el servicio:

curl http://127.0.0.1:8000/health

La documentación interactiva de la API está disponible en http://127.0.0.1:8000/docs.

Pasarela de SMS

Recepción de SMS

Registra un webhook en el panel de textbee.dev apuntando a https://<tu-host>/webhook/textbee, suscríbete al evento MESSAGE_RECEIVED y establece el secreto de firma como TEXTBEE_WEBHOOK_SECRET.

Cada entrega se verifica con HMAC-SHA256 sobre el cuerpo JSON sin procesar (cabecera X-Signature), se deduplica por el id de mensaje de la pasarela y se almacena en SQLite (se crean un usuario y una conversación si es necesario). El webhook responde inmediatamente y el mensaje se procesa en segundo plano.

Pruébalo localmente con una petición firmada simulada:

python scripts/mock_webhook.py "test_1" "What is 12 * 8?"

Pipeline de segundo plano

El webhook pone en cola cada mensaje entrante; un worker (iniciado por el ciclo de vida de FastAPI) lo procesa:

webhook -> store -> queue -> worker -> AI agent (+ MCP tools) -> reply
       -> TextBee send -> delivery status -> durable dedupe marker

• Almacenar primero, responder rápido: el webhook confirma el mensaje y lo pone en cola.

• Procesamiento asíncrono: una asyncio.Queue en proceso con WORKER_COUNT workers.

• Remitente correcto: la respuesta se envía solo al número de teléfono del mensaje entrante.

• Estado de entrega: las filas salientes registran status (sent/failed) y delivery_status (accepted/error).

• Reintentos: los fallos de envío de TextBee se reintentan hasta MAX_RETRIES con retroceso (backoff).

• Prevención de duplicados: el id de mensaje de la pasarela es único, y un marcador ProcessedMessage hace que el procesamiento sea idempotente entre reinicios.

• Fallos controlados: los fallos de IA/MCP se registran; los fallos de TextBee dejan la respuesta almacenada con status=failed en lugar de provocar un fallo del sistema.

Ejecuta un worker independiente sin el servidor web:

python -m app.workers.worker

Prueba de extremo a extremo sin un teléfono real

Simula el lado de envío de TextBee para que no se envíe ningún SMS real, pero el pipeline completo se ejecute:

# Terminal 1 - fake TextBee API (logs each accepted send)
python scripts/mock_textbee_api.py --port 9001

# Terminal 2 - demo MCP server (optional, for tool use)
python -m app.mcp.echo_server --port 8001

# Terminal 3 - the app pointing TextBee at the mock
set "TEXTBEE_API_URL=http://127.0.0.1:9001"
set "TEXTBEE_API_KEY=test-key"
set "TEXTBEE_WEBHOOK_SECRET=test_secret_at_least_20_chars"
set "MCP_SERVER_URL=http://127.0.0.1:8001/mcp"
uvicorn app.main:app --port 8000

# Terminal 4 - send a mock inbound SMS
python scripts/mock_webhook.py "e2e_1" "Hi there!"

A continuación, inspecciona database/sms_agent.db: el mensaje entrante tiene status=sent, la respuesta saliente tiene delivery_status=accepted, y processed_messages contiene el id de la pasarela.

Envío de SMS

# Preview the request without sending (no API key needed)
python -m app.sms.send_test +15551234567 "Hello" --dry-run

# Send for real (requires TEXTBEE_API_KEY in .env)
python -m app.sms.send_test +15551234567 "Hello"

IA local (Ollama)

El agente es independiente de la pasarela de SMS: lee el historial de conversaciones de SQLite y genera respuestas con el modelo de Ollama configurado.

1. Instala Ollama (https://ollama.com) y ejecútalo.

2. Descarga un modelo: ollama pull llama3.2.

3. Establece MODEL_NAME en .env (por defecto llama3.2).

El generador de respuestas:

• Detecta el idioma/escritura del mensaje entrante (urdu, árabe, cirílico, devanagari, CJK, etc.) e indica al modelo que responda en el mismo idioma.

• Construye el contexto a partir de los últimos MAX_CONTEXT_MESSAGES mensajes en SQLite.

• Almacena la respuesta del asistente como mensaje saliente.

• Elimina artefactos del modelo (p. ej. un eco inicial de assistant) y recorta a MAX_RESPONSE_CHARS para SMS.

• Agota el tiempo de espera después de AI_TIMEOUT_SECONDS y lanza OllamaError en caso de fallo.

Pruébalo en vivo:

python scripts/demo_agent.py        # English demo
python scripts/demo_agent_urdu.py   # Urdu language-matching demo

Herramientas MCP (bucle del agente)

El agente es un bucle de decisión:

message -> LLM -> needs a tool? -> MCP tool -> tool result -> LLM -> final answer

Cuando MCP_SERVER_URL está establecido, el agente descubre las herramientas del servidor, las pasa al modelo y ejecuta cualquier llamada a herramientas que solicite. Los resultados de las herramientas se retroalimentan hasta que el modelo produce una respuesta de texto final. El bucle está limitado a MAX_TOOL_CALLS para que un modelo con comportamiento incorrecto no pueda repetirse indefinidamente. Cada llamada a una herramienta se registra en la tabla tool_calls.

El repositorio incluye un pequeño servidor MCP de demostración (echo + calculadora) construido con el SDK oficial mcp:

# Terminal 1 - start the MCP server (Streamable HTTP)
python -m app.mcp.echo_server --port 8001

# Terminal 2 - run the full agent flow with real LLM + MCP tool
python scripts/demo_agent_tools.py "What is 17 * 23?"

Apunta MCP_SERVER_URL=http://127.0.0.1:8001/mcp en .env (o pasa la URL directamente a MCPClient).

Seguridad y endurecimiento

• Autenticación del webhook: verificación HMAC-SHA256 de X-Signature (tiempo constante).

• Validación de entrada: números de teléfono E.164, límite de longitud de mensaje, eliminación de caracteres de control.

• Inyección de prompts: el contenido del usuario se sanitiza ([instrucción bloqueada]) antes de llegar al modelo; el prompt del sistema prohíbe revelar secretos.

• Aislamiento por usuario: las respuestas se envían solo al remitente entrante; la búsqueda de conversación está limitada a la fila de usuario del remitente.

• Límite de velocidad: ventana deslizante por remitente en el webhook.

• Protección contra duplicados: gateway_message_id único + marcador ProcessedMessage persistente.

• Tiempos de espera: Ollama (AI_TIMEOUT_SECONDS), MCP (MCP_TIMEOUT_SECONDS), TextBee (TEXTBEE_TIMEOUT_SECONDS).

• Reintentos: reintentos de envío de TextBee con retroceso (MAX_RETRIES).

• Longitud de SMS: detección de GSM-7 (153/parte) frente a UCS-2 (67/parte) con división multiparte segura a nivel de grafema (app/sms/encoding.py).

• Copias de seguridad: instantáneas de copia de seguridad en línea de SQLite al inicio/apagado + database/backups/.

• Apagado controlado: el ciclo de vida detiene los workers y vacía la cola.

• Secretos: todos en .env (ignorado por git, chmod 600); nunca se registran.

Pruebas

pytest

Las pruebas usan una base de datos SQLite en memoria y simulan las APIs HTTP de TextBee y Ollama; las pruebas del cliente MCP usan el transporte en proceso del SDK. Las pruebas del pipeline cubren el flujo completo webhook -> cola -> agente -> envío -> estado, incluidos reintentos, deduplicación y discrepancia de remitente. Las pruebas de integración se ejecutan contra servidores reales de Ollama / MCP cuando son accesibles (se omiten en caso contrario):

python -m pytest tests/test_integration_ollama.py -v
python -m pytest tests/test_integration_agent_tools.py -v   # needs the MCP server on :8001

Estructura del proyecto

app/
├── main.py        # FastAPI app, lifespan (starts workers), /health
├── config.py      # Environment-driven settings
├── database.py    # SQLAlchemy engine + session
├── models.py      # users, conversations, messages, tool_calls, processed_messages
├── sms/
│   ├── webhook.py # MESSAGE_RECEIVED webhook: verify, validate, store, enqueue
│   ├── textbee.py # TextBee httpx client (send-sms) + signature verification
│   ├── pipeline.py# async queue + worker: agent -> split -> send -> status
│   ├── encoding.py# GSM-7 vs UCS-2 detection + grapheme-safe splitting
│   └── send_test.py  # CLI helper to send a test SMS
├── security.py  # input validation, prompt-injection, rate limiting
├── backup.py    # SQLite online backups
├── agent/
│   ├── agent.py    # agent decision loop: LLM <-> MCP tools, history, storage
│   ├── llm.py      # Ollama client (chat + tool calling, timeout, errors)
│   ├── prompts.py  # system prompt + message builder
│   ├── language.py # script-based language detection
│   └── memory.py   # conversation history helpers
├── mcp/
│   ├── client.py       # MCP client wrapper (connect, list, call)
│   └── echo_server.py  # demo MCP server (echo + calculate tools)
└── workers/       # standalone worker entry point