Telegram AI Agent 搭建教程：Python + OpenAI API + Docker 实现自己的 AI 机器人

一、AI Agent 正在从独立应用进入即时通讯工具

过去，我们想使用 AI 助手，通常需要打开 ChatGPT、Claude、Gemini 或其他专门的 AI 应用。

但对于很多日常任务而言，用户可能只需要一个简单的聊天入口。

例如：

“帮我总结这段文字。”

“查询一下网站现在是否正常。”

“今天有什么重要的 AI 模型更新？”

“帮我整理明天需要完成的工作。”

这些事情不一定需要一个复杂的网页管理后台。

如果直接通过 Telegram 等即时通讯工具发送消息，AI 就能够处理任务并返回结果，整个使用过程会更加自然。

这就是消息型 AI Agent（Messaging AI Agent）的基本思路。

2026 年，消息型 Agent 正逐渐受到关注。

它们与普通聊天机器人最大的区别，并不是聊天界面发生变化，而是背后可以连接：

大语言模型、外部工具、用户授权数据和长期任务系统。

本文将从零开始，使用 Telegram、Python、OpenAI API 和 Docker 搭建一个可以实际部署的基础版本。

需要说明，第一阶段实现的是具有消息收发和模型调用能力的私人 AI 助手。后续再加入受控工具调用、任务状态管理和审批机制，才能逐步形成真正能够执行复杂任务的 Agent。

二、为什么选择 Telegram？

Telegram 提供了相对完善的 Bot API。

开发者不需要自己实现整个聊天系统，就可以创建机器人并接收用户消息。

Telegram Bot API 支持：

• 文本消息；
• 命令；
• 图片与文件；
• Webhook；
• 按钮与交互控件；
• 消息发送；
• 群组与频道集成。

对于开发者来说，最重要的是：

Telegram 提供官方可用的机器人创建和消息接入机制。

而类似微信个人账号自动化的方案，可能受到平台接口、账号授权和使用规则的限制。

因此，本文选择 Telegram 作为教学示例。

这些架构思路后续也可以应用到具有正式开发者接口的其他消息平台。

三、先理解整个系统的工作流程

一个基础消息型 AI 助手包含四个主要组件。

第一，Telegram。

负责提供聊天入口。

第二，Webhook 服务。

接收 Telegram 发送的消息事件。

第三，AI API。

负责生成文本回复。

第四，消息发送服务。

把模型的回答返回 Telegram 聊天窗口。

整个流程可以概括为：

用户发送消息 → Telegram Bot API → Webhook → Python FastAPI → OpenAI Responses API → 返回模型结果 → Telegram sendMessage → 用户收到回复。

如果部署在自己的 VPS 上，还可以增加 Nginx 反向代理和 Docker Compose。

推荐架构：

Telegram → HTTPS Nginx → FastAPI Container → AI Provider。

这种方式能够让 Telegram 访问公开 HTTPS 地址，同时把真正的 Python 应用限制在服务器本地端口。

四、准备部署环境

本教程需要：

如果已经拥有一台部署了 Docker、Nginx 和网站管理面板的 VPS，也可以直接利用现有服务器。

初次测试不需要准备数据库。

但如果后续需要长期保存会话、定时任务或多用户记录，可以考虑 Redis 与 PostgreSQL。

五、创建 Telegram Bot

第一步：找到 BotFather

在 Telegram 中搜索官方机器人：

@BotFather

BotFather 是 Telegram 提供的机器人管理入口。

第二步：创建机器人

发送命令：

/newbot

BotFather 会要求输入机器人名称和用户名。

例如：

显示名称：

JieAGI Assistant

用户名：

JieAGI_Helper_Bot

注意用户名必须满足 Telegram 的命名规则，并以 bot 结尾。

第三步：保存 Token

创建成功后，BotFather 会返回 Bot Token。

它通常包含数字 ID 和后续的密钥字符串。

这个 Token 相当于机器人的控制凭据。

不要将真实 Token 发布在网页、GitHub 仓库或者技术文章中。

后续代码统一从环境变量读取。

六、获取自己的 Telegram Chat ID

为了避免其他人找到你的机器人后随意消耗 API 额度，建议初期只允许自己使用。

因此需要知道自己的 Telegram Chat ID。

在机器人聊天窗口中发送：

/start

首次设置 Webhook 之前，可以通过 Telegram Bot API 的 getUpdates 获取测试消息。

例如：

curl "https://api.telegram.org/bot${TG_BOT_TOKEN}/getUpdates"

返回结果中的私聊消息会包含类似以下字段：

{
  "message": {
    "chat": {
      "id": 123456789,
      "type": "private"
    }
  }
}

将 chat.id 保存下来。

请注意，getUpdates 与 Webhook 接收模式不能同时使用。配置 Webhook 后，就应该改为通过 Webhook 接收消息。

七、创建 Python 项目

新建一个项目目录：

mkdir telegram-ai-agent
cd telegram-ai-agent

推荐目录结构：

telegram-ai-agent/
├── app.py
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
└── .env

其中：

app.py 是主要业务逻辑。

requirements.txt 保存 Python 依赖。

Dockerfile 构建容器镜像。

docker-compose.yml 管理容器。

.env 保存 API Key、机器人 Token 和访问控制配置。

注意不要把 .env 提交到公开仓库。

八、安装 Python 依赖

在 requirements.txt 中写入：

fastapi
uvicorn[standard]
openai
httpx
python-dotenv

然后安装：

pip install -r requirements.txt

其中：

FastAPI 用于接收 Telegram Webhook。

Uvicorn 用于启动 HTTP 服务。

OpenAI SDK 用于访问模型。

HTTPX 用于向 Telegram API 发送回复。

python-dotenv 用于在本地开发时读取 .env 环境变量。

正式生产部署应将依赖版本锁定为已验证的版本，避免未来自动升级造成不兼容。

九、配置环境变量

创建 .env 文件：

TG_BOT_TOKEN=替换为TelegramBotToken

TG_WEBHOOK_SECRET=替换为随机生成的Webhook密钥

OPENAI_API_KEY=替换为OpenAIAPIKey

OPENAI_MODEL=gpt-6-luna

ALLOWED_CHAT_IDS=123456789

其中 ALLOWED_CHAT_IDS 填写前面获取的 Chat ID。

Webhook Secret 推荐使用随机字符串生成：

python -c "import secrets; print(secrets.token_urlsafe(32))"

生成的字符串可用于 Webhook 验证。

模型方面，本教程使用 OpenAI 的 gpt-6-luna。

根据 OpenAI 官方文档，该模型面向成本敏感的高频任务，并支持 Responses API。

需要注意，ChatGPT 网页订阅和 OpenAI API 计费属于不同体系，使用 API 可能产生独立费用。

十、编写完整的 Telegram AI 助手

在 app.py 中写入以下代码：

import os
import hmac
import logging

import httpx
from dotenv import load_dotenv
from fastapi import (
    FastAPI,
    Request,
    Header,
    BackgroundTasks,
    HTTPException,
)
from openai import AsyncOpenAI

load_dotenv()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("telegram-agent")

BOT_TOKEN = os.environ["TG_BOT_TOKEN"]
WEBHOOK_SECRET = os.environ["TG_WEBHOOK_SECRET"]
MODEL = os.getenv("OPENAI_MODEL", "gpt-6-luna")

ALLOWED_CHAT_IDS = {
    int(value.strip())
    for value in os.getenv(
        "ALLOWED_CHAT_IDS", ""
    ).split(",")
    if value.strip()
}

if not ALLOWED_CHAT_IDS:
    raise RuntimeError(
        "ALLOWED_CHAT_IDS must be configured"
    )

if not WEBHOOK_SECRET:
    raise RuntimeError(
        "TG_WEBHOOK_SECRET is required"
    )

app = FastAPI()
ai = AsyncOpenAI()

TELEGRAM_API = (
    f"https://api.telegram.org/bot{BOT_TOKEN}"
)

async def send_message(chat_id: int, text: str):
    async with httpx.AsyncClient(
        timeout=20.0
    ) as client:
        response = await client.post(
            f"{TELEGRAM_API}/sendMessage",
            json={
                "chat_id": chat_id,
                "text": text[:3500],
            },
        )
        response.raise_for_status()

async def process_message(
    chat_id: int,
    user_text: str
):
    if user_text.startswith("/start"):
        await send_message(
            chat_id,
            "你好！我是 JieAGI 私人 AI 助手。"
            "直接发送文字即可开始提问。"
        )
        return

    try:
        response = await ai.responses.create(
            model=MODEL,
            instructions=(
                "你是一名中文 AI 助手。"
                "回答应准确、清晰、简洁。"
                "不要编造实时新闻或数据。"
                "没有联网工具时要说明限制。"
            ),
            input=user_text,
            max_output_tokens=1200,
            store=False,
        )

        answer = (
            response.output_text
            or "本次没有生成文本结果。"
        )

        await send_message(chat_id, answer)

    except Exception:
        logger.exception(
            "AI task failed for chat %s",
            chat_id
        )

        try:
            await send_message(
                chat_id,
                "当前任务处理失败，请稍后重试。"
            )
        except Exception:
            logger.exception(
                "Failed to deliver error message"
            )

@app.post("/tg/webhook")
async def telegram_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
    x_telegram_bot_api_secret_token: str | None = Header(
        default=None
    ),
):
    received = (
        x_telegram_bot_api_secret_token or ""
    )

    if not hmac.compare_digest(
        received,
        WEBHOOK_SECRET
    ):
        raise HTTPException(
            status_code=403,
            detail="Forbidden"
        )

    update = await request.json()

    message = update.get("message") or {}
    chat = message.get("chat") or {}

    if chat.get("type") != "private":
        return {"ok": True}

    chat_id = chat.get("id")

    if chat_id not in ALLOWED_CHAT_IDS:
        return {"ok": True}

    text = message.get("text")

    if not isinstance(text, str):
        return {"ok": True}

    text = text.strip()

    if not text or len(text) > 4000:
        return {"ok": True}

    background_tasks.add_task(
        process_message,
        chat_id,
        text
    )

    return {"ok": True}

@app.get("/health")
async def health():
    return {"status": "ok"}

这段代码完成了基础功能：

接收 Telegram 消息。

检查 Webhook Secret。

限制只有指定 Chat ID 可以使用。

调用 OpenAI API 生成回答。

通过 Telegram sendMessage 返回内容。

代码还使用了 FastAPI BackgroundTasks，在接收 Webhook 后尽快返回确认，避免让 Telegram 一直等待完整模型回复。

需要注意，这属于单实例教学版，不是已经实现完整可靠队列的生产系统。

BackgroundTasks 运行在应用进程内，如果容器在任务执行途中重启，任务可能丢失。

此外，示例未实现持久化消息去重、分布式限流、多轮会话记忆和完整监控。

正式对外开放之前，应增加 Redis 或消息队列、按 update_id 去重、任务状态持久化与调用费用限制。

十一、先在本地启动服务

执行：

uvicorn app:app \
  --host 127.0.0.1 \
  --port 8000

检查：

http://127.0.0.1:8000/health

如果返回：

{
  "status": "ok"
}

说明 FastAPI 服务已经启动。

不过，此时 Telegram 仍然无法直接访问本地 127.0.0.1 地址。

接下来需要准备 HTTPS 公网入口。

十二、使用 Docker Compose 部署

先创建 Dockerfile：

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install \
    --no-cache-dir \
    -r requirements.txt

COPY app.py .

RUN useradd -u 10001 -m appuser

USER appuser

EXPOSE 8000

CMD [
  "uvicorn",
  "app:app",
  "--host",
  "0.0.0.0",
  "--port",
  "8000"
]

Dockerfile 里的多行 CMD 在实际保存时应写成合法的单行 JSON 数组：

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

再创建 docker-compose.yml：

services:
  telegram-agent:
    build: .
    container_name: telegram-ai-agent

    restart: unless-stopped

    ports:
      - "127.0.0.1:8018:8000"

    env_file:
      - .env

    read_only: true

    tmpfs:
      - /tmp

    security_opt:
      - no-new-privileges:true

    pids_limit: 128

    mem_limit: 512m

    logging:
      driver: json-file
      options:
        max-size: "10m"
        max-file: "3"

这里特意将服务绑定到：

127.0.0.1:8018

而不是直接向公网开放容器端口。

随后执行：

docker compose up -d --build

查看日志：

docker compose logs -f

再在服务器上检查：

curl http://127.0.0.1:8018/health

如果可以正常返回状态，说明容器已经启动。

十三、Nginx 反向代理配置

假设你准备使用：

https://bot.example.com

作为 Telegram Webhook 域名。

可以在对应 HTTPS 站点的 Nginx 配置中增加：

location = /tg/webhook {
    proxy_pass http://127.0.0.1:8018;

    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

    proxy_connect_timeout 10s;
    proxy_send_timeout 30s;
    proxy_read_timeout 30s;

    client_max_body_size 1m;
}

然后由网站现有的 HTTPS 配置处理 SSL 证书。

注意不要把示例域名直接当成真实地址。

正式部署时，需要替换成自己已经解析到服务器、并配置有效证书的域名。

十四、正式注册 Telegram Webhook

确认公网 HTTPS 地址可以访问后，向 Telegram 注册 Webhook：

curl -X POST \
  "https://api.telegram.org/bot${TG_BOT_TOKEN}/setWebhook" \
  -d "url=https://bot.example.com/tg/webhook" \
  -d "secret_token=${TG_WEBHOOK_SECRET}" \
  -d 'allowed_updates=["message"]'

Telegram 会将新的消息事件通过 HTTPS POST 请求发送到指定地址。

这里设置的 secret_token 会在请求头中以：

X-Telegram-Bot-Api-Secret-Token

发送。

Python 服务会检查这个值是否与本地环境变量一致。

官方也明确说明，配置 Webhook 后不能再同时使用 getUpdates 轮询模式。

因此注册完成后，应使用 getWebhookInfo 排查推送状态，而不是继续用 getUpdates 收消息。

十五、实际测试机器人

打开 Telegram，找到自己的 Bot。

发送：

你好，请介绍一下你自己。

如果全部配置正常，消息将经过：

Telegram → Webhook → FastAPI → OpenAI API。

随后获得 AI 回复。

你还可以测试：

请解释什么是 AI Agent。

或者：

帮我写一个 Python 列表去重函数。

需要注意，当前教程默认没有给模型联网搜索工具。

因此，如果问它实时新闻，模型可能无法提供可靠的最新结果。

不能因为它能回复消息，就认为它自动具备实时检索或其他工具执行能力。

十六、如何升级成真正能执行任务的 Agent？

完成基础消息助手后，可以逐步增加外部工具。

例如：

网站状态检查

用户输入：

/status

系统通过预先配置的只读监控接口获取网站状态。

然后把结果返回 Telegram。

AI 新闻助手

用户输入：

/news

Agent 调用已经授权的网页搜索服务，收集相关新闻。

随后结合发布时间、官方来源和文章价值整理结果。

服务器运维助手

用户输入：

/server

系统可以通过只读监控接口获取 CPU、内存、磁盘、容器状态等信息。

对于重启容器、修改配置或部署服务等操作，应增加明确的权限与人工确认。

项目开发助手

用户可以给 Agent 发送开发任务。

由 Agent 将任务委派给独立的 Coding 工具，并跟踪执行状态。

需要强调：

消息入口只是交互层，真正的 Agent 能力来自经过授权的工具调用和工作流控制。

不能直接把用户发送的任意自然语言转成 Shell 命令在服务器上执行。

十七、如何加入多轮对话和记忆？

当前示例每条消息都是独立请求。

如果需要让机器人记住上一轮聊天，就需要增加会话管理。

可以使用 Redis 保存一定数量的近期对话。

例如：

chat_id
  ├── message_1
  ├── message_2
  ├── message_3
  └── latest_summary

每次模型调用时，将当前问题与必要的历史信息一起发送。

为了控制成本，建议限制历史长度。

同时明确会话数据的保留时间。

如果涉及个人数据或企业敏感信息，还需要提供必要的访问控制和删除机制。

长期记忆不应把所有收到的内容直接写入永久数据库。

尤其是外部网页或工具返回内容，不应该自动成为可信的长期系统指令。

十八、如何避免 Telegram Bot 被恶意刷 API？

这是将机器人部署到公网前必须考虑的问题。

即使使用了 Chat ID 白名单，也需要考虑凭据泄露和错误配置。

建议至少增加：

用户白名单。 只允许明确授权的 Chat ID。

请求频率限制。 每个用户每分钟只能发起一定数量的请求。

Token 预算。 为单次任务设置最大输出和整体使用额度。

并发限制。 避免同时处理过多模型请求。

消息长度限制。 拒绝明显超出需要的输入。

更新去重。 按 Telegram update_id 防止重试消息被重复处理。

日志脱敏。 不记录真实 API Key、Bot Token 和不必要的消息正文。

费用告警。 在 API 服务商处设置预算或使用限制。

生产环境还应使用 Redis 或数据库保存消息处理状态，避免容器重启、Webhook 重试或者多副本部署时发生重复调用。

十九、消息型 AI Agent 与普通网页聊天有什么区别？

对于私人助手、日常运营提醒和简单数据查询，消息型入口通常比较方便。

对于复杂文件操作、数据分析和大量交互式界面，完整 Web 应用可能更加适合。

两者并不冲突。

一个成熟 Agent 系统完全可以同时支持 Telegram 和 Web 管理后台。

二十、常见问题

Telegram Bot 免费吗？

Telegram 提供机器人开发接口，但 AI 模型、VPS、网络与其他外部服务可能产生费用。

可以使用自己的 API 中转服务吗？

如果中转服务正确实现了所需的 OpenAI Responses API 协议，可以调整 SDK 的 base_url 和模型 ID。

但 Chat Completions 兼容并不自动意味着 Responses API 完全兼容，需要实际测试。

可以使用 DeepSeek 或 Qwen 吗？

可以根据这些模型服务商提供的 API 进行适配。

需要注意不同 Provider 的认证、请求参数和返回结构可能存在差异。

Telegram Bot 可以自动管理 Docker 吗？

技术上可以连接受控的容器管理接口，但不建议直接暴露宿主机 Docker Socket。

对于重启、删除、部署等操作，应采用专用管理服务、最小权限和人工审批。

可以部署在国内服务器吗？

需要确认服务器能够稳定访问 Telegram Bot API 和所选模型服务。网络可达性与服务合规要求应在部署前验证。

能不能接入微信？

取决于具体账号类型、官方接口能力和适用的平台规则，不应默认认为个人微信账号具有与 Telegram Bot 相同的开放机器人接口。

二十一、总结

消息型 AI Agent 的价值，在于把 AI 能力融入用户已经熟悉的交流工具中。

通过 Telegram Bot、Python、FastAPI、OpenAI API 和 Docker，我们可以搭建一个能够接收消息、调用模型并返回回答的私人 AI 助手。

在此基础上，再逐步加入：

搜索、记忆、任务队列、工具调用、权限控制、费用管理与人工审批。

一个简单的消息机器人，就可以逐渐发展为实际有用的 AI 工作助手。

但真正成熟的 Agent，并不取决于接入了多少工具。

而在于它能否持续、可靠、安全地完成用户明确授权的任务。

官方参考资料

1. Telegram — Bot API https://core.telegram.org/bots/api

2. Telegram — Bot Features https://core.telegram.org/bots/features

3. Telegram — From BotFather to Hello World https://core.telegram.org/bots/tutorial

4. OpenAI — Responses API https://developers.openai.com/api/docs/

5. OpenAI — GPT-6 Luna Model https://developers.openai.com/api/docs/models/gpt-6-luna