造一个 MCP 轮子发 reddit 图文 post
本来是想用github 上已有的服务的,但是都是需要创建应用,然后生成密钥来跑,很好但不是很适合。
所以用浏览器自动化的方式造了一个轮子用。
原理是先开一个浏览器的 remote debugging 端口,再使用自动化工具操作浏览器。
第一步开启浏览器服务
"C:Program FilesGoogleChromeApplicationchrome.exe" --user-data-dir="E:ChromeUserData" --remote-debugging-port=9922
第二部创建基础方法
保存图片
def save_img(url: str):
"""
Docstring for save_img
:param url: Description
:type url: str
"""
img_ext = Path((urlparse(url).path)).suffix
if not img_ext:
# 兼容没有后缀的图片 url
img_ext = ".jpg"
img_name = md5(url.encode('utf-8')).hexdigest() + img_ext
try:
urlretrieve(url, img_name)
img_full_path = os.path.abspath(img_name)
return img_full_path
except Exception as e:
print("使用 urlretrive 下载图片失败", str(e), "n")
return False
填充标题
def set_title(text: str):
"""
Docstring for fill_title
:param text: Description
:type text: str
"""
## 输入标题 ##
tab = browser.get_tab(url="reddit.com")
if not tab:
return False
try:
title_ele = tab.ele("@id=post-composer__title")
# print(title_ele.raw_text)
tab.actions.move_to(title_ele).click().type(text)
return True
except Exception as e:
print("⚠️ 输入标题异常 ->", str(e))
return False
上传图片
def set_image(img_path: str):
"""
Docstring for set_image
:param img_path: Description
:type img_path: str
"""
tab = browser.get_tab(url="reddit.com")
if not tab:
return False
if not os.path.exists(img_path):
print("图片文件不存在 n")
return False
try:
tab.set.upload_files(img_path)
media = tab.ele("@id=post-composer_media", timeout=10)
shadow = media.shadow_root
upload_btn = shadow.ele("@id=device-upload-button", timeout=10)
upload_btn.click()
tab.wait.upload_paths_inputted()
except Exception as e:
print("⚠️ 上传图片异常 ->", str(e))
return False
发布内容
def create_post(title: str, img_url: str):
# tab = get_reddit_tab()
img_path = save_img(url=img_url)
if not img_path:
print("图片下载失败了")
return False
tab = browser.get_tab(url="reddit.com")
if not tab:
return False
# https://www.reddit.com/r/whisky/submit/?type=IMAGE
tab.get("https://www.reddit.com/r/whisky/submit/?type=IMAGE")
## 输入标题 ##
title_filled = set_title(text=title)
if not title_filled:
print("标题输入异常了")
return False
## 上传图片 ##
image_uploaded = set_image(img_path=img_path)
if not image_uploaded:
print("图片上传异常了")
return False
## 点击发布 ##
posted = click_post()
if not posted:
print("发布按钮点击异常了")
return False
return "内容 post 成功了"
当然,不能忘了导包和连接浏览器
from urllib.request import urlretrieve
from hashlib import md5
from pathlib import Path
from urllib.parse import urlparse
import os
import time
from DrissionPage import Chromium
browser = Chromium("127.0.0.1:9922")
封装 MCP 服务
@mcp.tool()
async def create_reddit_post(title: str, img_url: str) -> List[TextContent]:
"""
发布 Reddit 图片帖。
Args:
title: Reddit 帖子标题
img_url: 图片 URL
"""
async with post_lock:
try:
if not title or not title.strip():
return [TextContent(type="text", text="❌ 发布失败:title 不能为空")]
if not img_url or not img_url.startswith(("http://", "https://")):
return [TextContent(type="text", text="❌ 发布失败:img_url 必须是 http 或 https 图片地址")]
result = await asyncio.to_thread(create_post, title.strip(), img_url.strip())
if result:
text = (
"✅ Reddit 图片帖发布完成nn"
f"标题:{title}n"
f"图片:{img_url}n"
f"结果:{result}"
)
else:
text = (
"❌ Reddit 图片帖发布失败nn"
f"标题:{title}n"
f"图片:{img_url}"
)
return [TextContent(type="text", text=text)]
except Exception as e:
return [TextContent(type="text", text=f"❌ 发布 Reddit 图片帖时发生异常:{str(e)}")]
@mcp.tool()
async def check_reddit_browser_status() -> List[TextContent]:
"""
检查 Reddit 浏览器状态。
"""
try:
tab = browser.get_tab(url="reddit.com")
if not tab:
text = "⚠️ 没有找到 reddit.com 标签页,请先打开 Reddit 页面。"
else:
text = f"✅ Reddit 浏览器状态正常nn当前标签页:{tab.url}"
return [TextContent(type="text", text=text)]
except Exception as e:
return [TextContent(type="text", text=f"❌ 检查浏览器状态失败:{str(e)}")]
if __name__ == "__main__":
print("=" * 50)
print("🚀 Reddit 图片帖发布 MCP Server 启动")
print("=" * 50)
print("服务模式:streamable-http")
print("服务地址:http://0.0.0.0:8800")
print("MCP 端点:http://127.0.0.1:8800/mcp")
print("")
print("可用工具:")
print("1. create_reddit_post")
print("2. check_reddit_browser_status")
print("=" * 50)
mcp.run(transport="streamable-http")
完事儿收工~
