AI Agent实时信息获取:基于Reddit API与MCP协议的last30days-skill项目实战
1. 项目概述:当AI Agent学会“冲浪”
最近在AI Agent的开发者圈子里,一个名为 last30days-skill 的项目火了,在GitHub上迅速斩获了超过44.3K的Star。这个数字背后,反映的是一个非常具体且迫切的需求:如何让我们的AI助手不再局限于静态的知识库,而是能像人类一样,主动去“逛”社区,获取最新、最鲜活的信息? last30days-skill 给出的答案,就是赋予AI Agent一项“搜索过去30天Reddit热门内容”的技能。
简单来说,你可以把它理解为一个专为AI Agent设计的、高度定制化的“社区热点雷达”。传统的AI应用,无论是基于GPT的聊天机器人还是各类自动化助手,其知识往往存在滞后性。它们可能精通历史事件、经典理论,但对“过去一周某个技术社区在热议什么框架”、“昨天游戏圈对某个新补丁的评价如何”这类动态信息却无能为力。 last30days-skill 正是为了解决这个“信息时效性”痛点而生。它通过封装对Reddit API的调用和智能解析,让AI Agent能够按需搜索并总结指定子版块(subreddit)在过去30天内的热门帖子,并将结构化的结果(如标题、链接、摘要、热度)反馈给Agent,从而极大地扩展了Agent的认知边界和实时辅助能力。
这个项目适合所有正在或计划开发AI Agent的开发者、产品经理以及对AI应用落地感兴趣的技术爱好者。无论你是想做一个能追踪科技动态的资讯助手,一个能分析市场情绪的舆情机器人,还是一个能随时解答最新游戏攻略的智能客服, last30days-skill 提供的基础能力都能让你事半功倍。它不仅仅是一个工具,更是一种思路的启发:AI Agent的未来,必然是与动态、实时的互联网数据进行深度交互。
2. 核心设计思路与架构拆解
2.1 为什么是Reddit?为什么是“过去30天”?
在构思一个让AI获取实时信息的技能时,数据源的选择至关重要。 last30days-skill 选择了Reddit,这背后有深刻的考量。首先,Reddit是一个基于兴趣社区的巨型论坛,内容包罗万象,从编程(r/programming)到日常生活(r/AskReddit),几乎每个垂直领域都有活跃的子版块。其次,Reddit的投票机制(赞/踩)天然地对内容进行了初步的质量筛选和热度排序,这比直接爬取无排序的论坛或社交媒体流要高效得多。最后,Reddit提供了相对友好和稳定的API,便于程序化访问。
而将时间范围限定在“过去30天”,则是一个平衡了“新鲜度”与“信息密度”的聪明设计。对于新闻、技术趋势、产品反馈这类信息,一个月的时间窗口既能覆盖大多数热点事件的完整生命周期(从爆发到沉淀),又能有效过滤掉过于陈旧的历史信息。同时,从API调用和数据处理的角度看,30天的数据量对于一次查询来说是可控的,不会给模型上下文窗口带来过大压力,也符合大多数免费或基础版API的调用限制。这个设计体现了项目作者对实际应用场景的深刻理解:AI Agent需要的是“近期热点”,而非历史档案。
2.2 技能(Skill)的标准化接口:MCP协议
last30days-skill 并非一个孤立的脚本,它的强大之处在于其遵循了Model Context Protocol(MCP)。你可以把MCP理解为AI Agent领域的“USB协议”。它为工具(Tools)或技能(Skills)与AI模型(如Claude、GPT)之间提供了一套标准化的通信方式。一个遵循MCP的技能,可以像插件一样,被任何支持该协议的AI Agent平台或框架(如Claude Desktop、Cline)轻松识别和调用。
具体到 last30days-skill ,它通过MCP向AI Agent暴露了几个核心的“工具函数”,例如 search_last_30_days 。当用户在聊天界面中向AI提出“看看最近r/machinelearning上有什么有趣的讨论”时,AI模型会理解用户的意图,并决定调用 last30days-skill 提供的这个搜索工具。模型会将必要的参数(如子版块名称、搜索关键词、返回结果数量)通过MCP格式传递给技能,技能执行完Reddit API查询和数据清洗后,再将结构化的结果通过MCP返回给模型,最后由模型组织成自然语言回复给用户。这种架构实现了AI核心逻辑与外部工具的解耦,使得技能的开发、部署和复用变得非常清晰和高效。
2.3 技术栈与工作流剖析
从技术实现上看, last30days-skill 是一个典型的Python服务。其核心工作流可以分解为以下几个步骤:
- 请求接收与解析 :技能作为一个HTTP服务器(通常使用FastAPI或类似框架),持续监听来自AI Agent(通过MCP服务器)的请求。收到包含搜索参数的请求后,首先进行参数验证和标准化。
- API调用与认证 :使用Python的
praw库或asyncpraw库(用于异步操作)与Reddit API进行交互。这里需要一个Reddit开发者账号并创建应用,以获取client_id,client_secret和user_agent进行OAuth2认证。技能使用这些凭证安全地访问Reddit数据。 - 数据获取与过滤 :向Reddit API发起请求,获取指定子版块按“热度”或“新帖”排序的帖子列表。然后,在本地对结果进行时间过滤,只保留过去30天内发布的帖子。这一步可能直接在API查询参数中完成(如果API支持时间范围筛选),也可能在获取数据后在内存中处理。
- 内容提取与结构化 :从每个符合条件的帖子中,提取关键字段:标题(title)、正文文本(selftext)、发布时间(created_utc)、得分(score)、评论数(num_comments)、永久链接(permalink)等。对于正文过长的帖子,可能需要调用LLM进行摘要总结,以节省后续模型的Token消耗。
- 结果格式化与返回 :将提取出的信息组装成MCP协议规定的响应格式(通常是JSON),返回给调用的AI Agent。这个响应结构清晰,便于AI模型理解和进一步加工。
注意 :在实际部署中,频繁调用Reddit API需注意速率限制。Reddit API对不同类型的请求(如应用类型、认证方式)有明确的每分钟/每日调用次数限制。生产环境中,需要实现请求队列、缓存机制(例如对同一子版块的查询结果缓存5-10分钟)和优雅降级策略,以避免触发限流导致服务不可用。
3. 核心细节解析与实操要点
3.1 Reddit API申请与配置避坑指南
要让 last30days-skill 跑起来,第一步也是最多新手卡住的一步:正确配置Reddit API。这个过程虽然不复杂,但细节决定成败。
首先,访问 https://www.reddit.com/prefs/apps 并用你的Reddit账号登录。点击页面底部的“Create App”或“Create Another App”。在创建表单中,你需要做出几个关键选择:
- name :你的应用名称,用户授权时会看到,可以随意起,比如“My AI Agent Scout”。
- type :这里必须选择 “script” 。对于
last30days-skill这种代表一个特定用户(你的机器人账号)进行后台操作的应用,“script”类型是最简单直接的。它使用用户名/密码的OAuth2流程,获取的token具有该账号的所有权限。 - redirect uri :对于“script”类型,可以填写
http://localhost:8080或任何一个有效的URI,实际上在简单的脚本认证中可能不会用到,但必须填写一个,不能为空。 - description :可选,填写应用描述。
创建成功后,你会看到应用信息面板,其中最关键的三样东西是: client_id (在应用名称下方,一串14位的字符)、 client_secret (一串27位的密钥),以及你登录Reddit的 username 和 password 。
实操心得 :很多人在这一步失败,是因为使用了错误的“应用类型”。如果你选择了“web app”或“installed app”,认证流程会复杂很多,需要处理OAuth2的回调。对于自用的AI技能,坚持用“script”类型能省去大量麻烦。另外,请务必妥善保管
client_secret,它相当于你的应用密码。
3.2 搜索策略与结果排序的权衡
last30days-skill 的核心是搜索,但“搜索”本身就有多种策略。直接使用Reddit API的 /r/{subreddit}/hot 或 /r/{subreddit}/new 端点获取列表然后本地过滤时间,是最简单的方式。但这种方式可能无法精准匹配用户查询中的关键词。
更高级的实现,会结合Reddit的搜索API。例如,使用 /r/{subreddit}/search 端点,并传入参数 q={keyword}&sort=relevance&t=month 。这样可以直接搜索过去一个月内相关度最高的帖子。两种策略各有优劣:
- 列表+过滤 :优点是可以确保获取到该子版块最热或最新的帖子,覆盖面广,即使没有关键词也能返回有价值的热门内容。缺点是如果用户提供了具体关键词,匹配精度可能不够。
- 直接搜索 :优点是结果与关键词高度相关。缺点是如果关键词比较泛或拼写有误,可能返回空结果,且会错过那些虽然没包含关键词但正在热议的相关话题。
一个健壮的 last30days-skill 实现应该结合两者。例如,默认情况下使用“列表+过滤”来获取热点概览;当用户明确提供了搜索词时,则切换到搜索模式。同时,可以对搜索结果进行二次排序,综合“相关性得分”和“帖子热度(得分)”,给AI Agent提供一份更优质的列表。
3.3 结果处理与Token经济
AI模型的上下文窗口(Token数)是宝贵的资源。一个子版块过去30天的热门帖子可能多达上百条,每条帖子还有标题、正文和评论。如果全部原样塞给AI,不仅会瞬间耗尽Token预算,还会让模型陷入信息过载,无法提炼重点。
因此,在将结果返回给AI Agent之前,进行智能化的结果处理至关重要。这包括:
- 数量限制 :严格控制返回的帖子数量,例如只返回前10条最相关或最热的帖子。
- 内容截断与摘要 :对于帖子正文,不能无脑全部返回。可以设定一个阈值(如500字符),超过部分进行截断,并标注“(内容已截断)”。更优的方案是,引入一个轻量级的文本摘要模型(或调用一次成本较低的LLM API),为长文生成一个2-3句话的摘要。这样用极少的Token就传达了核心信息。
- 结构化字段选择 :并非所有字段都有必要返回。对于AI生成回复而言,
title(标题)、score(热度)、num_comments(讨论度)、created_utc(时间)、permalink(链接)和一个简短的summary(摘要或正文片段)通常是核心字段。像作者、奖牌数量等信息可以酌情舍弃。
这种处理方式,本质上是在信息丰富度和Token消耗之间寻找最佳平衡点,是开发高效AI技能必须掌握的“经济学”。
4. 实操过程:从零搭建并集成到AI Agent
4.1 本地开发环境搭建与技能部署
假设我们使用Python进行开发。首先,创建一个新的项目目录并初始化环境。
# 创建项目目录
mkdir last30days-skill && cd last30days-skill
# 创建虚拟环境(推荐)
python -m venv venv
# 激活虚拟环境
# Windows: venv\Scripts\activate
# macOS/Linux: source venv/bin/activate
# 安装核心依赖
pip install mcp asyncpraw fastapi uvicorn pydantic
接下来,创建项目的核心文件结构。一个最简化的MCP技能通常包含一个主服务器文件(如 server.py )和一个声明工具的文件(如 tools.py )。
# server.py
import asyncpraw
from mcp.server import Server
from mcp.server.models import InitializationOptions
import asyncio
from typing import Any
# 导入我们将要定义的工具
from tools import search_last_30_days_tool
# 初始化MCP服务器
server = Server("last30days-skill")
# 注册工具
server.list_tools()(lambda: [search_last_30_days_tool])
server.call_tool()(search_last_30_days_tool.call)
async def main():
# 从环境变量读取Reddit API配置,更安全
reddit = asyncpraw.Reddit(
client_id="YOUR_CLIENT_ID",
client_secret="YOUR_CLIENT_SECRET",
user_agent="my_ai_agent_scout/0.1 by YourUsername",
username="YOUR_REDDIT_USERNAME", # 对于script类型需要
password="YOUR_REDDIT_PASSWORD",
)
# 可以将reddit客户端存储在server上下文或全局变量中,供工具函数使用
# 这里为了简单,我们作为示例。生产环境建议用依赖注入。
print("Last30Days Skill Server starting...")
async with server.run_stdio() as stream:
await stream.wait_closed()
if __name__ == "__main__":
asyncio.run(main())
# tools.py
from mcp.server.models import Tool
from pydantic import BaseModel, Field
import asyncpraw
from datetime import datetime, timedelta
import asyncio
from typing import List, Optional
# 定义工具的输入参数模型
class SearchLast30DaysInput(BaseModel):
subreddit: str = Field(description="要搜索的Reddit子版块名称,例如 'machinelearning', 'python'")
keyword: Optional[str] = Field(default=None, description="可选的搜索关键词。如果为空,则返回该子版块的热门帖子。")
limit: int = Field(default=10, ge=1, le=25, description="返回的帖子数量,默认为10,最大25")
# 定义工具函数
async def search_last_30_days(subreddit: str, keyword: Optional[str] = None, limit: int = 10) -> str:
"""
搜索指定Reddit子版块在过去30天内的帖子。
"""
# 注意:这里需要访问asyncpraw的reddit客户端实例。
# 在实际MCP服务器中,需要通过某种方式(如闭包、类属性、依赖注入)传递进来。
# 此处为函数逻辑示例。
reddit = get_reddit_client() # 假设这是一个获取已认证客户端的方法
try:
subreddit_obj = await reddit.subreddit(subreddit)
posts = []
cutoff_time = datetime.utcnow() - timedelta(days=30)
if keyword:
# 使用搜索功能
async for submission in subreddit_obj.search(
query=keyword, sort='relevance', time_filter='month', limit=limit
):
if datetime.utcfromtimestamp(submission.created_utc) > cutoff_time:
posts.append({
'title': submission.title,
'score': submission.score,
'num_comments': submission.num_comments,
'created': datetime.utcfromtimestamp(submission.created_utc).strftime('%Y-%m-%d'),
'url': f"https://reddit.com{submission.permalink}",
'summary': submission.selftext[:200] + '...' if len(submission.selftext) > 200 else submission.selftext
})
else:
# 获取热门帖子
async for submission in subreddit_obj.hot(limit=limit*2): # 多取一些,因为要过滤时间
post_time = datetime.utcfromtimestamp(submission.created_utc)
if post_time > cutoff_time:
posts.append({
'title': submission.title,
'score': submission.score,
'num_comments': submission.num_comments,
'created': post_time.strftime('%Y-%m-%d'),
'url': f"https://reddit.com{submission.permalink}",
'summary': submission.selftext[:200] + '...' if len(submission.selftext) > 200 else submission.selftext
})
if len(posts) >= limit:
break
if not posts:
return f"在 r/{subreddit} 中未找到过去30天内符合条件的热门帖子。"
# 格式化输出
result_lines = [f"在 r/{subreddit} 找到过去30天内的 {len(posts)} 个帖子:"]
for i, post in enumerate(posts, 1):
result_lines.append(
f"{i}. **{post['title']}** (热度: {post['score']}, 评论: {post['num_comments']}, 发布于: {post['created']})\n"
f" 摘要: {post['summary']}\n"
f" 链接: {post['url']}"
)
return "\n\n".join(result_lines)
except Exception as e:
return f"搜索过程中发生错误: {str(e)}"
# 将函数包装成MCP工具
search_last_30_days_tool = Tool(
name="search_last_30_days",
description="搜索指定Reddit子版块在过去30天内的热门或相关帖子。",
inputSchema=SearchLast30DaysInput.model_json_schema(),
callback=search_last_30_days, # 注意:这里需要适配MCP服务器的调用方式,实际可能更复杂
)
注意 :以上代码为高度简化的示例,旨在说明逻辑。真实的MCP工具注册和异步客户端管理会更复杂,需要参考
mcpPython SDK的官方文档。关键点在于理解工具的定义、参数验证和与Reddit API的交互流程。
4.2 与Claude Desktop或Cline集成
部署好技能服务器后,下一步是让它被AI Agent使用。以目前流行的Claude Desktop为例,你需要编辑其配置文件来添加这个自定义MCP服务器。
找到Claude Desktop的配置文件(通常在 ~/.config/Claude/claude_desktop_config.json 或类似路径)。在 mcpServers 部分添加一个新的服务器配置:
{
"mcpServers": {
"last30days-skill": {
"command": "/path/to/your/venv/bin/python",
"args": ["/full/path/to/your/last30days-skill/server.py"],
"env": {
"REDDIT_CLIENT_ID": "your_client_id",
"REDDIT_CLIENT_SECRET": "your_client_secret",
"REDDIT_USERNAME": "your_username",
"REDDIT_PASSWORD": "your_password"
}
}
}
}
配置完成后,重启Claude Desktop。在聊天界面中,Claude现在应该就能识别并使用 search_last_30_days 这个工具了。你可以直接输入:“用 last30days-skill 查一下 r/programming 最近有什么热点。” Claude会自动调用该工具并返回格式化的结果。
对于像Cline这样的代码编辑器AI Agent插件,集成方式类似,通常也是通过编辑其设置文件,指定MCP服务器的启动命令和环境变量。
4.3 配置优化与性能调优
在本地开发时,上述配置可以工作。但对于持续运行或希望分享给他人的技能,需要更稳健的配置。
- 环境变量管理 :绝对不要将API密钥硬编码在代码中。使用
python-dotenv库从.env文件读取,或通过容器环境变量注入。上面Claude配置中的env部分就是一种方式。 - 错误处理与重试 :网络请求和API调用总会失败。必须在代码中为
asyncpraw的调用添加全面的错误处理(try-except),并对可重试的错误(如网络超时、速率限制)实现指数退避重试机制。 - 请求缓存 :为了避免对同一子版块的重复查询在短时间内耗尽API限额,可以引入一个简单的内存缓存(如
cachetools库的TTLCache)或外部缓存(如Redis)。例如,将(subreddit, keyword)作为键,查询结果作为值,缓存5-10分钟。 - 日志记录 :添加详细的日志记录(使用
logging模块),记录每次工具调用、API请求和发生的错误,这对于后期调试和监控至关重要。 - 异步优化 :确保整个处理链是异步的,从接收MCP请求到调用Reddit API,避免阻塞事件循环,这对于高并发场景尤为重要。
5. 常见问题与排查技巧实录
在实际开发和集成 last30days-skill 的过程中,你几乎一定会遇到下面这些问题。这里记录了我的踩坑实录和解决方案。
5.1 认证失败与API限流
- 问题现象 :技能启动失败,或运行时抛出
prawcore.exceptions.ResponseException,提示Invalid Client、Invalid Grant或403 Forbidden、429 Too Many Requests。 - 排查思路 :
- 检查凭证 :这是最常见的问题。逐字核对
client_id,client_secret,username,password和user_agent。确保没有多余的空格,密码中的特殊字符是否正确转义。user_agent格式建议为“<平台>:<应用名>:<版本号> (by /u/<你的Reddit用户名>)”。 - 检查应用类型 :再次确认在Reddit开发者面板创建的应用类型是 “script” ,而不是其他类型。
- 检查速率限制 :Reddit API对“script”类型应用有明确的速率限制(通常每分钟60次请求)。如果你的技能被频繁调用,很容易触发限流。查看错误信息是否包含
429状态码。
- 检查凭证 :这是最常见的问题。逐字核对
- 解决方案 :
- 对于凭证错误,重新生成
client_secret或检查账号密码。 - 对于速率限制,必须实现请求缓存。例如,对完全相同的查询(子版块+关键词),在短时间内返回缓存结果。可以使用
cachetools.TTLCache(maxsize=100, ttl=300)实现一个5分钟过期的内存缓存。 - 在代码中捕获
429异常,并实现指数退避重试逻辑,例如等待(2 ** retry_count)秒后再重试。
- 对于凭证错误,重新生成
5.2 MCP服务器连接失败或工具不可见
- 问题现象 :Claude Desktop重启后,在聊天中尝试使用技能,Claude回复说“不知道这个工具”或直接没有反应。查看Claude日志可能看到连接错误。
- 排查思路 :
- 检查配置文件路径 :确保Claude配置文件中
command和args指向的Python解释器和脚本路径是 绝对路径 ,并且完全正确。虚拟环境中的Python路径尤其要注意。 - 检查服务器启动 :手动在终端运行你配置的启动命令(例如
/path/to/venv/bin/python /path/to/server.py),看技能服务器是否能正常启动并打印出启动日志,而不是立刻报错退出。 - 检查标准I/O :MCP服务器通过标准输入输出(stdio)与主机通信。确保你的服务器代码正确使用了
mcp.server的run_stdio()方法,并且没有其他打印输出干扰了协议通信。调试时,可以先注释掉所有非必要的print语句。 - 检查工具注册 :确认在服务器代码中正确注册了工具列表和回调函数。工具的名称(
name)和描述(description)是AI模型识别它的关键。
- 检查配置文件路径 :确保Claude配置文件中
- 解决方案 :
- 使用绝对路径,并确保Claude Desktop有权限执行该命令。
- 在技能服务器代码开始时添加详细的日志,记录启动步骤和工具注册情况,便于排查。
- 参考MCP官方提供的示例服务器代码,确保协议握手和通信流程正确。
5.3 搜索结果不理想或为空
- 问题现象 :工具能调用,但返回的结果很少,或者完全不相关,甚至经常返回“未找到帖子”。
- 排查思路 :
- 子版块名称 :确认子版块名称拼写正确,且是公开版块。有些版块是私密的,需要加入才能访问。
- 时间过滤逻辑 :检查代码中计算
cutoff_time(30天前)的逻辑是否正确。注意datetime.utcnow()和帖子时间戳created_utc都是UTC时间。 - 搜索策略 :如果使用了关键词搜索,尝试在Reddit网站上手动用同样的关键词和子版块搜索,对比结果。Reddit的搜索算法可能无法匹配太复杂或太长的短语。
- API端点限制 :
/hot端点返回的帖子数量可能有限(通常是前几百个)。如果这个子版块过去30天非常活跃,一些稍早的热帖可能已经不在/hot列表里了。可以尝试结合/new和/hot,或者使用搜索API并设置sort=top&t=month来获取月度热门帖。
- 解决方案 :
- 对于活跃度一般的版块,使用
subreddit.hot(limit=100)通常足够。 - 对于非常活跃的版块(如 r/funny),考虑增加获取数量(如
limit=200)再进行时间过滤,或者实现分页获取。 - 优化关键词搜索:对用户输入的关键词进行简单的预处理,如转小写、去除停用词,或者尝试将长句拆分成多个关键词进行组合搜索。
- 对于活跃度一般的版块,使用
5.4 技能响应慢或超时
- 问题现象 :AI调用技能后,需要等待很长时间才有回复,有时甚至超时。
- 排查思路 :
- 网络延迟 :访问Reddit API的延迟可能较高,尤其是在非北美地区。
- 同步阻塞 :代码中是否存在同步的、耗时的操作(如读写文件、复杂的CPU计算)阻塞了异步事件循环?确保所有I/O操作都是异步的(使用
async/await)。 - 获取数据过多 :是否一次性获取了过多的帖子或尝试获取了完整的帖子正文及大量评论?这会导致单次API响应数据量巨大,拉长响应时间。
- 解决方案 :
- 实现请求缓存,这是提升响应速度最有效的方法,对于重复查询,直接返回缓存结果。
- 严格限制单次返回的帖子数量(如10条),并对帖子正文进行强制摘要或截断。
- 检查代码,确保没有在工具函数中执行同步的HTTP请求或数据库查询。全部改用异步库。
- 考虑为技能设置一个合理的超时时间(例如10秒),并在超时后返回一个友好的错误信息,而不是让用户无限等待。
开发这样一个技能,从跑通Demo到稳定可用,最大的挑战往往不在核心逻辑,而在这些“周边”的工程化细节上。处理好认证、缓存、错误和性能,你的AI Agent才能真正可靠地“逛”起来。
更多推荐

所有评论(0)