Skip to main content
Glama
eneszh
mustafaansari4564

mcp-sms-agent

by mustafaansari4564
OverviewSchemaRelated ServersScoreDiscussions
Python
Remote

SMS AI 代理

一个独立的 AI 驱动的短信代理。入站短信通过 TextBee 短信网关到达，FastAPI webhook 将其交给本地 AI 代理，代理在需要时可以使用 MCP 工具，AI 的回复通过 TextBee 发送回同一号码。

状态： 完整流水线已连通并针对生产环境加固——TextBee webhook -> SQLite -> 异步后台工作器 -> 本地 AI (Ollama) -> MCP 工具 -> 短信感知回复拆分 -> TextBee 发送 -> 投递状态。

生产指南（部署、HTTPS、安全、备份、故障排查）： docs/PRODUCTION.md

架构

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

环境要求

  • Python 3.11+

  • SQLite（随 Python 捆绑）

安装

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

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

pip install -r requirements.txt

配置

复制示例环境文件并填入你的值：

cp .env.example .env

TextBee (https://textbee.dev) 设置：

变量

用途

TEXTBEE_API_URL

API 基础 URL（默认 https://api.textbee.dev）

TEXTBEE_API_KEY

仪表盘 API 密钥，作为 x-api-key 请求头发送

TEXTBEE_DEVICE_ID

发送所使用的设备；省略则使用默认设备

TEXTBEE_WEBHOOK_SECRET

你的 MESSAGE_RECEIVED webhook 订阅的签名密钥；用于验证 X-Signature 请求头（至少 20 个字符）

Ollama 设置：

变量

用途

OLLAMA_BASE_URL

Ollama 服务器 URL（默认 http://127.0.0.1:11434）

MODEL_NAME

用于生成回复的模型（例如 llama3.2；先用 ollama pull llama3.2 拉取）

AI_TIMEOUT_SECONDS

等待模型响应的最大秒数

MAX_CONTEXT_MESSAGES

作为对话上下文包含的最近消息数

MAX_RESPONSE_CHARS

回复最大长度（为短信保持简短）

MCP 设置：

变量

用途

MCP_SERVER_URL

MCP 服务器 URL（Streamable HTTP 传输）。留空 = 无工具

MCP_TIMEOUT_SECONDS

等待 MCP 连接/列出/调用的最大秒数

MAX_TOOL_CALLS

每轮对话的最大工具调用次数（循环保护）

SQLite 数据库文件在首次启动时自动创建于 database/sms_agent.db。

运行

uvicorn app.main:app --reload

然后检查服务：

curl http://127.0.0.1:8000/health

交互式 API 文档可在 http://127.0.0.1:8000/docs 查看。

短信网关

接收短信

在 textbee.dev 仪表盘中注册一个指向 https://<your-host>/webhook/textbee 的 webhook，订阅 MESSAGE_RECEIVED 事件，并将签名密钥设置为 TEXTBEE_WEBHOOK_SECRET。

每次投递都通过 HMAC-SHA256 对原始 JSON 请求体进行验证 （X-Signature 请求头），按网关消息 ID 去重，并存储在 SQLite 中（必要时自动创建用户和会话）。webhook 立即响应， 消息在后台处理。

用模拟签名请求在本地测试：

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

后台流水线

webhook 将每条入站消息入队；一个工作器（由 FastAPI 生命周期启动）处理它：

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

  • 先存储，快速响应：webhook 提交消息并入队。

  • 异步处理：进程内 asyncio.Queue，带 WORKER_COUNT 个工作器。

  • 正确的发送方：回复仅发送给入站消息的电话号码。

  • 投递状态：出站记录记录 status（sent/failed）和 delivery_status（accepted/error）。

  • 重试：TextBee 发送失败时最多重试 MAX_RETRIES 次，带退避。

  • 重复预防：网关消息 ID 唯一，ProcessedMessage 标记使处理在重启后保持幂等。

  • 优雅失败：AI/MCP 失败被记录；TextBee 失败将回复保留为 status=failed 而不是崩溃。

不启动 Web 服务器，独立运行工作器：

python -m app.workers.worker

无需真实手机的端到端测试

模拟 TextBee 发送端，这样不会发送真实短信，但整个流水线会运行：

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

然后检查 database/sms_agent.db：入站消息的 status=sent， 出站回复的 delivery_status=accepted，processed_messages 中包含网关 ID。

发送短信

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

本地 AI (Ollama)

代理独立于短信网关：它从 SQLite 读取会话历史， 并使用配置的 Ollama 模型生成回复。

  1. 安装 Ollama (https://ollama.com) 并启动它。

  2. 拉取模型：ollama pull llama3.2。

  3. 在 .env 中设置 MODEL_NAME（默认 llama3.2）。

回复生成器：

  • 检测入站消息的语言/文字（乌尔都语、阿拉伯语、西里尔字母、 天城文、中日韩等），并指示模型用相同语言回复。

  • 从 SQLite 中最近的 MAX_CONTEXT_MESSAGES 条消息构建上下文。

  • 将助手回复存储为出站消息。

  • 去除模型产物（例如开头的 assistant 回声）并裁剪到 MAX_RESPONSE_CHARS 以适配短信。

  • 在 AI_TIMEOUT_SECONDS 后超时，失败时抛出 OllamaError。

实时试用：

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

MCP 工具（代理循环）

代理是一个决策循环：

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

当设置了 MCP_SERVER_URL 时，代理发现服务器的工具，将其 传递给模型，并执行模型请求的任何工具调用。工具结果被 反馈回去，直到模型产生最终文本答案。循环上限为 MAX_TOOL_CALLS，这样行为异常的模型不会无限循环。每次工具调用都 记录在 tool_calls 表中。

该仓库包含一个用官方 mcp SDK 构建的微型演示 MCP 服务器（echo + 计算器）：

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

在 .env 中设置 MCP_SERVER_URL=http://127.0.0.1:8001/mcp（或将 URL 直接 传给 MCPClient）。

安全与加固

  • Webhook 认证：HMAC-SHA256 X-Signature 验证（恒定时间比较）。

  • 输入验证：E.164 电话号码、消息长度上限、控制字符 剥离。

  • 提示注入防护：用户内容在到达模型前被净化（[blocked instruction]）； 系统提示禁止泄露机密。

  • 按用户隔离：回复仅发送给入站发送者； 会话查找限定在发送者的用户行内。

  • 速率限制：webhook 上按发送者的滑动窗口。

  • 重复保护：唯一 gateway_message_id + 持久化 ProcessedMessage 标记。

  • 超时：Ollama（AI_TIMEOUT_SECONDS）、MCP（MCP_TIMEOUT_SECONDS）、 TextBee（TEXTBEE_TIMEOUT_SECONDS）。

  • 重试：TextBee 发送带退避重试（MAX_RETRIES）。

  • 短信长度：GSM-7（153/条）与 UCS-2（67/条）检测， 带字素安全的多部分拆分（app/sms/encoding.py）。

  • 备份：启动/关闭时的 SQLite 在线备份快照 + database/backups/。

  • 优雅关闭：生命周期停止工作器并排空队列。

  • 机密：全部在 .env 中（已 gitignore，chmod 600）；绝不记录日志。

测试

pytest

测试使用内存 SQLite 数据库，并模拟 TextBee 和 Ollama HTTP API；MCP 客户端测试使用 SDK 的进程内传输。流水线测试 覆盖完整的 webhook -> 队列 -> 代理 -> 发送 -> 状态流程， 包括重试、去重和发送者不匹配。集成测试在真实 Ollama / MCP 服务器可达时运行（否则跳过）：

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

项目结构

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

This server cannot be installed

-
license - not tested
Not graded
quality - not tested
C
maintenance

How are these scores calculated?

Maintenance

–Maintainers
–Response time
–Release cycle
–Releases (12mo)
Commit activity

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Related MCP Connectors

  • Hail

    Phone, SMS & email for AI agents — one remote MCP endpoint, OAuth login, zero install.

  • AgentPhone

    Give AI agents real phone numbers, messages, and voice calls via MCP.

  • anythingmcp

    Self-hosted MCP gateway: turn any API, database or MCP server into AI connectors — no code.

View all MCP Connectors

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/mustafaansari4564/mcp-sms-agent'

If you have feedback or need assistance with the MCP directory API, please join our Discord server