SMS AI Agent

A standalone AI-powered SMS agent. Incoming SMS arrive via the TextBee SMS gateway, a FastAPI webhook hands them to a local AI agent, which may use MCP tools when needed, and the AI's reply is sent back to the same number via TextBee.

Status: Full pipeline connected and hardened for production - TextBee webhook -> SQLite -> async background worker -> local AI (Ollama) -> MCP tools -> SMS-aware reply splitting -> TextBee send -> delivery status.

Production guide (deployment, HTTPS, security, backup, troubleshooting): docs/PRODUCTION.md

Architecture

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

Requirements

• Python 3.11+

• SQLite (bundled with Python)

Installation

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

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

pip install -r requirements.txt

Configuration

Copy the example environment file and fill in your values:

cp .env.example .env

TextBee (https://textbee.dev) settings:

Ollama settings:

MCP settings:

The SQLite database file is created automatically at database/sms_agent.db on first startup.

Run

uvicorn app.main:app --reload

Then check the service:

curl http://127.0.0.1:8000/health

Interactive API docs are available at http://127.0.0.1:8000/docs.

SMS gateway

Receiving SMS

Register a webhook in the textbee.dev dashboard pointing at https://<your-host>/webhook/textbee, subscribe to the MESSAGE_RECEIVED event, and set the signing secret as TEXTBEE_WEBHOOK_SECRET.

Each delivery is verified with HMAC-SHA256 over the raw JSON body (X-Signature header), deduplicated by the gateway message id, and stored in SQLite (a user and conversation are created as needed). The webhook responds immediately and the message is processed in the background.

Test it locally with a mock signed request:

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

Background pipeline

The webhook enqueues each inbound message; a worker (started by the FastAPI lifespan) processes it:

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

• Store first, respond fast: the webhook commits the message and enqueues.

• Async processing: an in-process asyncio.Queue with WORKER_COUNT workers.

• Correct sender: the reply is sent only to the phone number of the inbound message.

• Delivery status: outbound rows record status (sent/failed) and delivery_status (accepted/error).

• Retries: TextBee send failures retry up to MAX_RETRIES with backoff.

• Duplicate prevention: the gateway message id is unique, and a ProcessedMessage marker makes processing idempotent across restarts.

• Graceful failure: AI/MCP failures are logged; TextBee failures leave the reply stored with status=failed rather than crashing.

Run a standalone worker without the web server:

python -m app.workers.worker

End-to-end test without a real phone

Mock the TextBee send side so no real SMS is sent, but the whole pipeline runs:

# 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!"

Then inspect database/sms_agent.db: the inbound message has status=sent, the outbound reply has delivery_status=accepted, and processed_messages contains the gateway id.

Sending 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"

Local AI (Ollama)

The agent is independent of the SMS gateway: it reads conversation history from SQLite and generates replies with the configured Ollama model.

1. Install Ollama (https://ollama.com) and start it.

2. Pull a model: ollama pull llama3.2.

3. Set MODEL_NAME in .env (default llama3.2).

The reply generator:

• Detects the language/script of the incoming message (Urdu, Arabic, Cyrillic, Devanagari, CJK, etc.) and instructs the model to reply in the same language.

• Builds context from the last MAX_CONTEXT_MESSAGES messages in SQLite.

• Stores the assistant reply as an outbound message.

• Strips model artifacts (e.g. a leading assistant echo) and trims to MAX_RESPONSE_CHARS for SMS.

• Times out after AI_TIMEOUT_SECONDS and raises OllamaError on failure.

Try it live:

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

MCP tools (agent loop)

The agent is a decision loop:

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

When MCP_SERVER_URL is set, the agent discovers the server's tools, passes them to the model, and executes any tool calls it requests. Tool results are fed back until the model produces a final text answer. The loop is capped at MAX_TOOL_CALLS so a misbehaving model cannot loop forever. Every tool call is recorded in the tool_calls table.

The repo includes a tiny demo MCP server (echo + calculator) built with the official mcp SDK:

# 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?"

Point MCP_SERVER_URL=http://127.0.0.1:8001/mcp in .env (or pass the URL directly to MCPClient).

Security & hardening

• Webhook auth: HMAC-SHA256 X-Signature verification (constant-time).

• Input validation: E.164 phone numbers, message length cap, control-char stripping.

• Prompt injection: user content is sanitized ([blocked instruction]) before it reaches the model; the system prompt forbids revealing secrets.

• Per-user isolation: replies are sent only to the inbound sender; the conversation lookup is scoped to the sender's user row.

• Rate limiting: per-sender sliding window on the webhook.

• Duplicate protection: unique gateway_message_id + durable ProcessedMessage marker.

• Timeouts: Ollama (AI_TIMEOUT_SECONDS), MCP (MCP_TIMEOUT_SECONDS), TextBee (TEXTBEE_TIMEOUT_SECONDS).

• Retries: TextBee send retries with backoff (MAX_RETRIES).

• SMS length: GSM-7 (153/part) vs UCS-2 (67/part) detection with grapheme-safe multipart splitting (app/sms/encoding.py).

• Backups: SQLite online-backup snapshots at startup/shutdown + database/backups/.

• Graceful shutdown: the lifespan stops workers and drains the queue.

• Secrets: all in .env (gitignored, chmod 600); never logged.

Tests

pytest

Tests use an in-memory SQLite database and mock the TextBee and Ollama HTTP APIs; the MCP client tests use the SDK's in-process transport. The pipeline tests cover the full webhook -> queue -> agent -> send -> status flow, including retries, dedupe, and sender mismatch. Integration tests run against real Ollama / MCP servers when they are reachable (skipped otherwise):

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

Project layout

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