RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
755 lines
30 KiB
Python
755 lines
30 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
ros-mail-mcp (HTTP 版)
|
||
======================
|
||
部署在 bj 上，与主后端 mailWeb.py 同目录，直接 import 复用其工具函数。
|
||
|
||
架构：
|
||
agent (HTTPS) -> nginx api.rrrosmontis.icu/v1/mail/mcp/ -> 127.0.0.1:12006 (本服务)
|
||
|
||
认证双层：
|
||
1. HTTP 层（可选）：Bearer API Key（环境变量 MCP_API_KEY，设置才强制校验）；
|
||
2. 业务层（主要）：三种登录方式任选——
|
||
- login_sso：ROSMANKA ID SSO token 换取（推荐，agent 场景，密码不过 agent 侧）；
|
||
- login：邮箱密码换取一次性 token（复用 mailWeb 的 generate_token/validate_token，
|
||
UUID 随机、30 天滑动过期（每次使用自动续期）、仅存内存）；
|
||
其余工具只认 token；也兼容直接传 password（仅白名单域名，走 TLS）。
|
||
"""
|
||
import os
|
||
import re
|
||
import json
|
||
import time
|
||
import imaplib
|
||
import logging
|
||
|
||
import uvicorn
|
||
from starlette.types import ASGIApp, Receive, Scope, Send
|
||
from starlette.responses import JSONResponse
|
||
|
||
from fastmcp import FastMCP
|
||
|
||
# ---- 复用主后端 mailWeb.py 的工具函数（同目录）----
|
||
from mailWeb import (
|
||
imap_login,
|
||
smtp_login,
|
||
decode_mime_str,
|
||
decode_attachment_names,
|
||
extract_email_body,
|
||
build_cid_map,
|
||
replace_cid_with_data,
|
||
generate_token,
|
||
validate_token,
|
||
TOKEN_EXPIRE,
|
||
AUTODISCOVER_DOMAINS,
|
||
SSO_SECRETS_PATH,
|
||
_sso_service_token,
|
||
)
|
||
import email as email_mod
|
||
|
||
# ============================== 配置 ==============================
|
||
HOST = os.environ.get('MCP_HOST', '127.0.0.1')
|
||
PORT = int(os.environ.get('MCP_PORT', '12006'))
|
||
API_KEY = os.environ.get('MCP_API_KEY', '')
|
||
ALLOWED_DOMAINS = [d.lower() for d in AUTODISCOVER_DOMAINS] # 密码直传时的域名白名单
|
||
|
||
# ---- SSO 绑定（OAuth 式）配置 ----
|
||
OIDC_CLIENT_ID = 'mail-mcp'
|
||
OIDC_AUTHORIZE_URL = 'https://api.rrrosmontis.icu/v1/sso/oidc/authorize'
|
||
BIND_CALLBACK_URL = 'https://api.rrrosmontis.icu/v1/mail/mcp/bind/callback'
|
||
SSO_TOKEN_URL = 'http://127.0.0.1:4550/v1/sso/oidc/token'
|
||
SSO_EXCHANGE_OIDC_URL = 'http://127.0.0.1:4550/v1/sso/service/exchange_oidc'
|
||
_BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
BINDINGS_PATH = os.path.join(_BASE_DIR, 'mcp_accounts.json') # 绑定账户持久化（凭据 Fernet 加密）
|
||
BIND_KEY_PATH = os.path.join(_BASE_DIR, 'mcp_bind.key') # 加密密钥（自动生成，仅本机）
|
||
PENDING_BINDS = {} # 进行中的绑定请求（内存，10 分钟有效）：bind_id -> {state, created, result}
|
||
|
||
logging.basicConfig(level=logging.INFO,
|
||
format='%(asctime)s [mail-mcp] %(levelname)s %(message)s')
|
||
log = logging.getLogger('mail-mcp')
|
||
|
||
if not API_KEY:
|
||
log.warning('MCP_API_KEY 未设置！端点将不受 Bearer Key 保护，仅建议本地调试。')
|
||
|
||
mcp = FastMCP('ros-mail-mcp', instructions='Rosmontis 邮件 MCP。三种身份方式：① bind_account_start 发起 SSO 绑定'
|
||
'（OAuth 式，推荐：一次绑定后所有工具用 account=邮箱地址 调用，多账户重复绑定即可）；'
|
||
'② login_sso / login 换取 token（30 天有效，每次使用自动续期）；'
|
||
'③ 每次调用直传 password（仅限白名单域名）。')
|
||
|
||
|
||
# ============================== 绑定存储（Fernet 加密持久化） ==============================
|
||
_bind_fernet = None
|
||
|
||
|
||
def _get_bind_fernet():
|
||
global _bind_fernet
|
||
if _bind_fernet is None:
|
||
from cryptography.fernet import Fernet
|
||
if os.path.isfile(BIND_KEY_PATH):
|
||
with open(BIND_KEY_PATH, 'rb') as f:
|
||
key = f.read().strip()
|
||
else:
|
||
key = Fernet.generate_key()
|
||
with open(BIND_KEY_PATH, 'wb') as f:
|
||
f.write(key)
|
||
_bind_fernet = Fernet(key)
|
||
return _bind_fernet
|
||
|
||
|
||
def _load_bindings():
|
||
try:
|
||
with open(BINDINGS_PATH, 'r', encoding='utf-8') as f:
|
||
doc = json.load(f)
|
||
doc.setdefault('accounts', {})
|
||
return doc
|
||
except Exception:
|
||
return {'accounts': {}}
|
||
|
||
|
||
def _save_bindings(doc):
|
||
tmp = BINDINGS_PATH + '.tmp'
|
||
with open(tmp, 'w', encoding='utf-8') as f:
|
||
json.dump(doc, f, ensure_ascii=False, indent=2)
|
||
os.replace(tmp, BINDINGS_PATH)
|
||
|
||
|
||
def _get_bound_credentials(account: str):
|
||
"""按邮箱地址取绑定凭据并滑动刷新 last_used。返回 (email, password) 或 None。"""
|
||
doc = _load_bindings()
|
||
acc = doc['accounts'].get(account.lower())
|
||
if not acc:
|
||
return None
|
||
acc['last_used'] = time.time()
|
||
_save_bindings(doc)
|
||
secret = _get_bind_fernet().decrypt(acc['secret_enc'].encode('ascii')).decode('utf-8')
|
||
return acc['email'], secret
|
||
|
||
|
||
def _oidc_client_secret():
|
||
try:
|
||
with open(SSO_SECRETS_PATH, 'r', encoding='utf-8') as f:
|
||
return json.load(f).get('oidc_client_secrets', {}).get(OIDC_CLIENT_ID, '')
|
||
except Exception:
|
||
return ''
|
||
|
||
|
||
# ============================== 认证逻辑 ==============================
|
||
class AuthError(Exception):
|
||
pass
|
||
|
||
|
||
def _resolve_credentials(email: str, token: str, password: str, account: str = ''):
|
||
"""返回 (email_addr, password)。优先级：token > account（SSO 绑定）> password 直传。"""
|
||
email = (email or '').strip()
|
||
token = (token or '').strip()
|
||
account = (account or '').strip()
|
||
if token:
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
raise AuthError('token 无效或已过期（闲置超 30 天），请重新调用 login 工具')
|
||
tok_email, tok_pw = creds
|
||
if email and email.lower() != tok_email.lower():
|
||
raise AuthError('token 与该 email 不匹配')
|
||
return tok_email, tok_pw
|
||
if account:
|
||
creds = _get_bound_credentials(account)
|
||
if not creds:
|
||
raise AuthError(f'账户 {account} 未绑定，请先调用 bind_account_start 完成 SSO 绑定')
|
||
if email and email.lower() != creds[0].lower():
|
||
raise AuthError('email 与绑定的 account 不匹配')
|
||
return creds
|
||
if password and email:
|
||
domain = email.split('@')[-1].lower()
|
||
if domain not in ALLOWED_DOMAINS:
|
||
raise AuthError(f'该服务仅允许白名单域名邮箱（{"/".join(ALLOWED_DOMAINS)}）')
|
||
return email, password
|
||
raise AuthError('缺少凭据：请提供 token / account（SSO 绑定，推荐）/ email + password 之一')
|
||
|
||
|
||
def _find_sent_folder(conn):
|
||
typ, folder_data = conn.list()
|
||
if typ != 'OK':
|
||
return 'Sent'
|
||
candidates = ['Sent', 'Sent Messages', 'Sent Items', '已发送', '已发送邮件']
|
||
for line in folder_data:
|
||
line_str = line.decode() if isinstance(line, bytes) else line
|
||
m = re.search(r'"([^"]+)"$', line_str)
|
||
name = m.group(1) if m else (line_str.split()[-1].strip('"') if line_str.split() else None)
|
||
if not name:
|
||
continue
|
||
if any(name.lower() == c.lower() for c in candidates) or 'sent' in name.lower() or '已发送' in name:
|
||
return name
|
||
return 'Sent'
|
||
|
||
|
||
# ============================== MCP 工具 ==============================
|
||
@mcp.tool()
|
||
def login(email: str, password: str) -> dict:
|
||
"""登录邮箱，换取后续工具使用的访问 token（30 天有效，每次使用自动续期）。
|
||
|
||
密码只在本次 TLS 加密调用中传输，换取 UUID token 后不再使用。
|
||
这也是唯一需要密码的工具。
|
||
|
||
Args:
|
||
email: 完整邮箱地址
|
||
password: 邮箱密码
|
||
"""
|
||
conn = imap_login(email, password)
|
||
if not conn:
|
||
raise AuthError('邮箱登录失败：用户名或密码错误')
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
token = generate_token(email, password)
|
||
return {'token': token, 'email': email, 'expires_in': TOKEN_EXPIRE}
|
||
|
||
|
||
@mcp.tool()
|
||
def login_sso(sso_token: str) -> dict:
|
||
"""通过 ROSMANKA ID SSO 登录，换取后续工具使用的 MCP 访问 token（30 天有效，每次使用自动续期）。
|
||
|
||
无需邮箱密码：agent 从已登录 sso.rrrosmontis.icu 的用户处获得 SSO token 后调用本工具，
|
||
服务端用 service token 向 SSO 服务交换邮箱凭据并立即转为本服务的随机 token，
|
||
凭据和 SSO token 均不返回给 agent。
|
||
|
||
Args:
|
||
sso_token: ROSMANKA ID SSO 颁发的用户 token
|
||
"""
|
||
import requests as rq
|
||
from mailWeb import SSO_API_BASE, _sso_service_token
|
||
try:
|
||
r = rq.post(SSO_API_BASE + '/service/exchange',
|
||
json={'service': 'mail', 'token': sso_token.strip()},
|
||
headers={'X-SSO-Service-Token': _sso_service_token()}, timeout=10)
|
||
try:
|
||
d = r.json()
|
||
except Exception:
|
||
d = {}
|
||
except Exception:
|
||
raise RuntimeError('SSO 服务暂时不可用，请稍后重试')
|
||
if r.status_code != 200 or not d.get('ok'):
|
||
raise AuthError(d.get('error') or 'SSO token 无效或已过期，请重新登录 sso.rrrosmontis.icu')
|
||
email_addr, password = d['external_id'], d['external_secret']
|
||
conn = imap_login(email_addr, password)
|
||
if conn:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
token = generate_token(email_addr, password)
|
||
return {'token': token, 'email': email_addr, 'sso': True}
|
||
|
||
|
||
@mcp.tool()
|
||
def logout(token: str) -> dict:
|
||
"""注销 token，立即使其失效。
|
||
|
||
Args:
|
||
token: login 返回的 token
|
||
"""
|
||
from mailWeb import TOKEN_MAP
|
||
if token and token in TOKEN_MAP:
|
||
del TOKEN_MAP[token]
|
||
return {'success': True}
|
||
|
||
|
||
# ============================== SSO 绑定工具（OAuth 式） ==============================
|
||
@mcp.tool()
|
||
def bind_account_start() -> dict:
|
||
"""开始绑定一个 SSO 邮箱账户（OAuth 式流程，全程无需经手密码）。
|
||
|
||
返回一个 SSO 授权链接，请用户在浏览器打开并完成登录授权（绑定哪个 SSO 账户由用户在浏览器里决定，
|
||
重复调用即可绑定多个账户）。用户看到"绑定成功"页面后，调用 bind_account_status 确认。
|
||
|
||
Returns:
|
||
bind_id: 绑定标识（10 分钟内有效）
|
||
url: 需要用户在浏览器打开的 SSO 授权链接
|
||
"""
|
||
import secrets as _s
|
||
from urllib.parse import quote
|
||
state = _s.token_urlsafe(24)
|
||
bind_id = _s.token_hex(8)
|
||
now = time.time()
|
||
for k in [k for k, v in PENDING_BINDS.items() if now - v['created'] > 600]:
|
||
del PENDING_BINDS[k]
|
||
PENDING_BINDS[bind_id] = {'state': state, 'created': now, 'result': None}
|
||
url = (f"{OIDC_AUTHORIZE_URL}?client_id={OIDC_CLIENT_ID}"
|
||
f"&redirect_uri={quote(BIND_CALLBACK_URL, safe='')}"
|
||
f"&response_type=code&scope=openid%20email&prompt=login&state={state}")
|
||
return {'bind_id': bind_id, 'url': url, 'expires_in': 600,
|
||
'hint': '请把 url 发给用户，让其用浏览器打开完成 SSO 登录授权，然后调用 bind_account_status(bind_id) 确认'}
|
||
|
||
|
||
@mcp.tool()
|
||
def bind_account_status(bind_id: str) -> dict:
|
||
"""查询 SSO 绑定结果。用户在浏览器完成授权后返回 bound=true 与绑定的邮箱地址。
|
||
|
||
Args:
|
||
bind_id: bind_account_start 返回的绑定标识
|
||
"""
|
||
p = PENDING_BINDS.get(bind_id)
|
||
if not p:
|
||
raise ValueError('bind_id 不存在或已过期（10 分钟），请重新调用 bind_account_start')
|
||
if p['result']:
|
||
email_addr = p['result']['email']
|
||
del PENDING_BINDS[bind_id]
|
||
return {'bound': True, 'email': email_addr}
|
||
return {'bound': False, 'hint': '用户尚未完成授权，请等用户在浏览器完成后再查询'}
|
||
|
||
|
||
@mcp.tool()
|
||
def list_accounts() -> dict:
|
||
"""列出所有已通过 SSO 绑定的邮箱账户（不含任何凭据信息）。"""
|
||
doc = _load_bindings()
|
||
return {'accounts': [{'email': a['email'],
|
||
'bound_at': int(a.get('bound_at', 0)),
|
||
'last_used': int(a.get('last_used', 0))}
|
||
for a in doc['accounts'].values()]}
|
||
|
||
|
||
@mcp.tool()
|
||
def unbind_account(email: str, token: str) -> dict:
|
||
"""解绑一个已绑定的邮箱账户（删除服务端托管凭据，该邮箱此后需重新绑定或改用 token 登录）。
|
||
|
||
安全要求：必须提供该账户自己的有效 token（先 login 证明你是该账户的所有者），防止跨账户解绑。
|
||
|
||
Args:
|
||
email: 要解绑的邮箱地址
|
||
token: 该邮箱的有效 MCP token
|
||
"""
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
raise AuthError('token 无效或已过期')
|
||
if creds[0].lower() != (email or '').lower():
|
||
raise AuthError('token 与要解绑的 email 不匹配')
|
||
doc = _load_bindings()
|
||
if email.lower() in doc['accounts']:
|
||
del doc['accounts'][email.lower()]
|
||
_save_bindings(doc)
|
||
return {'success': True}
|
||
|
||
|
||
# ============================== SSO 绑定回调（浏览器回跳，非 MCP 工具） ==============================
|
||
@mcp.custom_route('/bind/callback', methods=['GET'])
|
||
async def bind_callback(request):
|
||
from starlette.responses import HTMLResponse
|
||
code = request.query_params.get('code', '')
|
||
state = request.query_params.get('state', '')
|
||
if not code or not state:
|
||
return HTMLResponse('<h3>绑定失败：缺少 code 或 state 参数</h3>', status_code=400)
|
||
target = None
|
||
for bid, p in PENDING_BINDS.items():
|
||
if p['state'] == state:
|
||
target = (bid, p)
|
||
break
|
||
if not target:
|
||
return HTMLResponse('<h3>绑定失败：授权请求不存在或已过期，请回到 Agent 重新发起绑定</h3>', status_code=400)
|
||
_, pend = target
|
||
import requests as rq
|
||
try:
|
||
tr = rq.post(SSO_TOKEN_URL, data={
|
||
'grant_type': 'authorization_code',
|
||
'code': code,
|
||
'client_id': OIDC_CLIENT_ID,
|
||
'client_secret': _oidc_client_secret(),
|
||
'redirect_uri': BIND_CALLBACK_URL,
|
||
}, timeout=10)
|
||
td = tr.json()
|
||
if tr.status_code != 200 or 'access_token' not in td:
|
||
return HTMLResponse(f'<h3>绑定失败：换取 access token 失败（{td.get("error", "unknown")}）</h3>',
|
||
status_code=502)
|
||
xr = rq.post(SSO_EXCHANGE_OIDC_URL,
|
||
json={'service': 'mail', 'access_token': td['access_token']},
|
||
headers={'X-SSO-Service-Token': _sso_service_token()}, timeout=10)
|
||
xd = xr.json()
|
||
if xr.status_code != 200 or not xd.get('ok'):
|
||
return HTMLResponse(f'<h3>绑定失败：{xd.get("error", "SSO 凭据交换失败")}</h3>', status_code=502)
|
||
except Exception as e:
|
||
return HTMLResponse(f'<h3>绑定失败：SSO 服务暂时不可用（{e}）</h3>', status_code=502)
|
||
|
||
email_addr = xd['external_id']
|
||
secret = xd['external_secret']
|
||
doc = _load_bindings()
|
||
doc['accounts'][email_addr.lower()] = {
|
||
'email': email_addr,
|
||
'secret_enc': _get_bind_fernet().encrypt(secret.encode('utf-8')).decode('ascii'),
|
||
'bound_at': time.time(),
|
||
'last_used': time.time(),
|
||
}
|
||
_save_bindings(doc)
|
||
pend['result'] = {'email': email_addr}
|
||
return HTMLResponse(
|
||
'<!DOCTYPE html><html lang="zh-CN"><head><meta charset="utf-8"><title>绑定成功</title></head>'
|
||
'<body style="font-family:sans-serif;text-align:center;padding-top:15vh">'
|
||
f'<h2>✅ 邮箱绑定成功</h2><p>已将 <b>{email_addr}</b> 绑定到邮件 MCP 服务。</p>'
|
||
'<p>你可以关闭本页，回到 Agent 继续。</p></body></html>')
|
||
|
||
|
||
@mcp.tool()
|
||
def list_mailboxes(email: str = '', token: str = '', password: str = '', account: str = '') -> dict:
|
||
"""列出 IMAP 邮箱中的所有文件夹。
|
||
|
||
Args:
|
||
email: 邮箱地址（用 token 时可省略，服务端自动取 token 对应邮箱）
|
||
token: login 返回的 token（推荐）
|
||
password: 邮箱密码直传（仅限白名单域名，不推荐）
|
||
account: 已 SSO 绑定的邮箱地址（推荐，bind_account_start 一次绑定后即用）
|
||
"""
|
||
email_addr, pw = _resolve_credentials(email, token, password, account)
|
||
conn = imap_login(email_addr, pw)
|
||
if not conn:
|
||
raise AuthError('IMAP 登录失败')
|
||
try:
|
||
typ, data = conn.list()
|
||
if typ != 'OK':
|
||
return {'mailboxes': []}
|
||
mailboxes = []
|
||
for line in data:
|
||
line_str = line.decode() if isinstance(line, bytes) else line
|
||
m = re.search(r'"([^"]+)"$', line_str)
|
||
if m:
|
||
mailboxes.append(m.group(1))
|
||
return {'mailboxes': mailboxes}
|
||
finally:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
@mcp.tool()
|
||
def list_messages(email: str = '', token: str = '', password: str = '', account: str = '',
|
||
folder: str = 'INBOX', page: int = 1, per_page: int = 20) -> dict:
|
||
"""分页列出指定文件夹中的邮件（默认 INBOX，按时间倒序，返回最新一页）。
|
||
|
||
Args:
|
||
email: 邮箱地址（用 token 时可省略）
|
||
token: login 返回的 token（推荐）
|
||
password: 邮箱密码直传（仅限白名单域名，不推荐）
|
||
account: 已 SSO 绑定的邮箱地址（推荐，bind_account_start 一次绑定后即用）
|
||
folder: 文件夹名，默认 INBOX
|
||
page: 页码，从 1 开始，默认 1
|
||
per_page: 每页数量，默认 20
|
||
"""
|
||
email_addr, pw = _resolve_credentials(email, token, password, account)
|
||
conn = imap_login(email_addr, pw)
|
||
if not conn:
|
||
raise AuthError('IMAP 登录失败')
|
||
try:
|
||
conn.select(folder)
|
||
result, data = conn.uid('search', None, 'ALL')
|
||
if result != 'OK':
|
||
return {'messages': [], 'has_more': False}
|
||
uid_list = data[0].split()
|
||
total = len(uid_list)
|
||
start = max(0, total - page * per_page)
|
||
end = total - (page - 1) * per_page
|
||
page_uids = uid_list[start:end] if start < end else []
|
||
messages = []
|
||
for uid in page_uids:
|
||
result, msg_data = conn.uid('fetch', uid, '(BODY.PEEK[HEADER])')
|
||
if result != 'OK' or not msg_data or not msg_data[0]:
|
||
continue
|
||
msg = email_mod.message_from_bytes(msg_data[0][1])
|
||
messages.append({
|
||
'uid': uid.decode() if isinstance(uid, bytes) else uid,
|
||
'from': decode_mime_str(msg.get('From', '')),
|
||
'subject': decode_mime_str(msg.get('Subject', '')),
|
||
'date': msg.get('Date', ''),
|
||
'attachments': decode_attachment_names(msg.get('X-Attachments', '')),
|
||
})
|
||
messages.reverse()
|
||
return {'messages': messages, 'has_more': page * per_page < total, 'total': total}
|
||
finally:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
@mcp.tool()
|
||
def list_sent(email: str = '', token: str = '', password: str = '', account: str = '',
|
||
page: int = 1, per_page: int = 20) -> dict:
|
||
"""分页列出已发送邮件（按时间倒序）。
|
||
|
||
Args:
|
||
email: 邮箱地址（用 token 时可省略）
|
||
token: login 返回的 token（推荐）
|
||
password: 邮箱密码直传（仅限白名单域名，不推荐）
|
||
account: 已 SSO 绑定的邮箱地址（推荐，bind_account_start 一次绑定后即用）
|
||
page: 页码，从 1 开始，默认 1
|
||
per_page: 每页数量，默认 20
|
||
"""
|
||
email_addr, pw = _resolve_credentials(email, token, password, account)
|
||
conn = imap_login(email_addr, pw)
|
||
if not conn:
|
||
raise AuthError('IMAP 登录失败')
|
||
try:
|
||
sent_folder = _find_sent_folder(conn)
|
||
try:
|
||
conn.select(sent_folder)
|
||
except Exception:
|
||
return {'messages': [], 'has_more': False}
|
||
result, data = conn.uid('search', None, 'ALL')
|
||
if result != 'OK':
|
||
return {'messages': [], 'has_more': False}
|
||
uid_list = data[0].split()
|
||
total = len(uid_list)
|
||
start = max(0, total - page * per_page)
|
||
end = total - (page - 1) * per_page
|
||
page_uids = uid_list[start:end] if start < end else []
|
||
messages = []
|
||
for uid in page_uids:
|
||
result, msg_data = conn.uid('fetch', uid, '(BODY.PEEK[HEADER])')
|
||
if result != 'OK' or not msg_data or not msg_data[0]:
|
||
continue
|
||
msg = email_mod.message_from_bytes(msg_data[0][1])
|
||
messages.append({
|
||
'uid': uid.decode() if isinstance(uid, bytes) else uid,
|
||
'from': decode_mime_str(msg.get('From', '')),
|
||
'to': decode_mime_str(msg.get('To', '')),
|
||
'subject': decode_mime_str(msg.get('Subject', '')),
|
||
'date': msg.get('Date', ''),
|
||
'attachments': decode_attachment_names(msg.get('X-Attachments', '')),
|
||
})
|
||
messages.reverse()
|
||
return {'messages': messages, 'has_more': page * per_page < total, 'total': total}
|
||
finally:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
@mcp.tool()
|
||
def read_message(email: str = '', token: str = '', password: str = '', account: str = '',
|
||
uid: str = '', folder: str = 'INBOX') -> dict:
|
||
"""读取指定 UID 的邮件全文（HTML 正文中的内嵌图片 cid: 引用会被替换为 data URL）。
|
||
|
||
Args:
|
||
email: 邮箱地址（用 token 时可省略）
|
||
token: login 返回的 token（推荐）
|
||
password: 邮箱密码直传（仅限白名单域名，不推荐）
|
||
account: 已 SSO 绑定的邮箱地址（推荐，bind_account_start 一次绑定后即用）
|
||
uid: 邮件 UID（从 list_messages / list_sent 获取）
|
||
folder: 文件夹名，默认 INBOX
|
||
"""
|
||
if not uid:
|
||
raise ValueError('uid 不能为空')
|
||
email_addr, pw = _resolve_credentials(email, token, password, account)
|
||
conn = imap_login(email_addr, pw)
|
||
if not conn:
|
||
raise AuthError('IMAP 登录失败')
|
||
try:
|
||
conn.select(folder)
|
||
result, data = conn.uid('fetch', uid, '(RFC822)')
|
||
if result != 'OK' or not data or not data[0]:
|
||
raise ValueError(f'邮件不存在: uid={uid}')
|
||
msg = email_mod.message_from_bytes(data[0][1])
|
||
body = extract_email_body(msg)
|
||
cid_map = build_cid_map(msg)
|
||
if cid_map:
|
||
body = replace_cid_with_data(body, cid_map)
|
||
return {
|
||
'uid': uid,
|
||
'body': body,
|
||
'subject': decode_mime_str(msg.get('Subject', '')),
|
||
'from': decode_mime_str(msg.get('From', '')),
|
||
'to': decode_mime_str(msg.get('To', '')),
|
||
'date': msg.get('Date', ''),
|
||
'attachments': decode_attachment_names(msg.get('X-Attachments', '')),
|
||
}
|
||
finally:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
@mcp.tool()
|
||
def read_attachment(email: str = '', token: str = '', password: str = '', account: str = '',
|
||
uid: str = '', filename: str = '', folder: str = 'INBOX') -> dict:
|
||
"""下载指定邮件中的附件（返回 base64 内容）。
|
||
|
||
建议仅在需要时调用；大附件会占用较多上下文。
|
||
|
||
Args:
|
||
email: 邮箱地址（用 token 时可省略）
|
||
token: login 返回的 token（推荐）
|
||
password: 邮箱密码直传（仅限白名单域名，不推荐）
|
||
account: 已 SSO 绑定的邮箱地址（推荐，bind_account_start 一次绑定后即用）
|
||
uid: 邮件 UID
|
||
filename: 附件文件名（从 list_messages 的 attachments 字段获取）
|
||
folder: 文件夹名，默认 INBOX
|
||
"""
|
||
import base64
|
||
if not uid or not filename:
|
||
raise ValueError('uid 和 filename 均不能为空')
|
||
email_addr, pw = _resolve_credentials(email, token, password, account)
|
||
conn = imap_login(email_addr, pw)
|
||
if not conn:
|
||
raise AuthError('IMAP 登录失败')
|
||
try:
|
||
conn.select(folder)
|
||
result, data = conn.uid('fetch', uid, '(RFC822)')
|
||
if result != 'OK' or not data or not data[0]:
|
||
raise ValueError(f'邮件不存在: uid={uid}')
|
||
msg = email_mod.message_from_bytes(data[0][1])
|
||
for part in msg.walk():
|
||
disp = part.get('Content-Disposition', '')
|
||
if not disp or 'attachment' not in disp:
|
||
continue
|
||
raw_name = part.get_filename()
|
||
if not raw_name:
|
||
continue
|
||
if decode_mime_str(raw_name) == filename:
|
||
payload = part.get_payload(decode=True)
|
||
if payload is None:
|
||
continue
|
||
content_type = part.get_content_type() or 'application/octet-stream'
|
||
return {
|
||
'uid': uid,
|
||
'filename': filename,
|
||
'content_type': content_type,
|
||
'size': len(payload),
|
||
'content_b64': base64.b64encode(payload).decode('ascii'),
|
||
}
|
||
raise ValueError(f'附件不存在: {filename}')
|
||
finally:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
@mcp.tool()
|
||
def send_email(email: str = '', token: str = '', password: str = '', account: str = '',
|
||
to: str = '', subject: str = '', body: str = '') -> dict:
|
||
"""发送 HTML 格式邮件，发送后自动保存到已发送文件夹。
|
||
|
||
Args:
|
||
email: 发件人邮箱地址（用 token 时可省略）
|
||
token: login 返回的 token（推荐）
|
||
password: 邮箱密码直传（仅限白名单域名，不推荐）
|
||
account: 已 SSO 绑定的邮箱地址（推荐，bind_account_start 一次绑定后即用）
|
||
to: 收件人邮箱地址
|
||
subject: 邮件主题
|
||
body: 邮件正文（HTML 格式）
|
||
"""
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
if not all([to, subject]):
|
||
raise ValueError('to 和 subject 不能为空')
|
||
email_addr, pw = _resolve_credentials(email, token, password, account)
|
||
smtp = smtp_login(email_addr, pw)
|
||
if not smtp:
|
||
raise AuthError('SMTP 登录失败')
|
||
try:
|
||
msg = MIMEMultipart()
|
||
msg['From'] = email_addr
|
||
msg['To'] = to
|
||
msg['Subject'] = subject
|
||
msg.attach(MIMEText(body or '', 'html', 'utf-8'))
|
||
smtp.send_message(msg)
|
||
smtp.quit()
|
||
except Exception as e:
|
||
raise RuntimeError(f'发送失败: {e}')
|
||
|
||
# 保存到已发送
|
||
try:
|
||
conn = imap_login(email_addr, pw)
|
||
if conn:
|
||
try:
|
||
sent_folder = _find_sent_folder(conn)
|
||
try:
|
||
conn.select(sent_folder)
|
||
except Exception:
|
||
try:
|
||
conn.create(sent_folder)
|
||
conn.select(sent_folder)
|
||
except Exception:
|
||
return {'success': True}
|
||
conn.append(sent_folder, '\\Seen',
|
||
imaplib.Time2Internaldate(time.time()), msg.as_bytes())
|
||
conn.close()
|
||
finally:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
return {'success': True}
|
||
|
||
|
||
@mcp.tool()
|
||
def delete_message(email: str = '', token: str = '', password: str = '', account: str = '',
|
||
uid: str = '', folder: str = 'INBOX') -> dict:
|
||
"""删除指定 UID 的邮件（移动到 \\Deleted 并 expunge，不可恢复）。
|
||
|
||
Args:
|
||
email: 邮箱地址（用 token 时可省略）
|
||
token: login 返回的 token（推荐）
|
||
password: 邮箱密码直传（仅限白名单域名，不推荐）
|
||
account: 已 SSO 绑定的邮箱地址（推荐，bind_account_start 一次绑定后即用）
|
||
uid: 邮件 UID
|
||
folder: 文件夹名，默认 INBOX
|
||
"""
|
||
if not uid:
|
||
raise ValueError('uid 不能为空')
|
||
email_addr, pw = _resolve_credentials(email, token, password, account)
|
||
conn = imap_login(email_addr, pw)
|
||
if not conn:
|
||
raise AuthError('IMAP 登录失败')
|
||
try:
|
||
conn.select(folder)
|
||
result, _ = conn.uid('store', uid, '+FLAGS', '\\Deleted')
|
||
if result != 'OK':
|
||
raise RuntimeError('删除失败')
|
||
conn.expunge()
|
||
return {'success': True}
|
||
finally:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ============================== HTTP 层（Bearer Key 中间件） ==============================
|
||
class BearerKeyMiddleware:
|
||
"""ASGI 中间件：设置了 MCP_API_KEY 时，除 /healthz 外所有请求必须携带
|
||
Authorization: Bearer <MCP_API_KEY>；未设置则不校验（依赖业务层 SSO/token 认证）。"""
|
||
|
||
def __init__(self, app: ASGIApp):
|
||
self.app = app
|
||
|
||
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
|
||
if scope['type'] != 'http':
|
||
await self.app(scope, receive, send)
|
||
return
|
||
path = scope.get('path', '')
|
||
if path == '/healthz':
|
||
resp = JSONResponse({'status': 'ok', 'service': 'ros-mail-mcp'})
|
||
await resp(scope, receive, send)
|
||
return
|
||
if path.startswith('/bind/'): # SSO 绑定回调来自用户浏览器，无 Bearer 头
|
||
await self.app(scope, receive, send)
|
||
return
|
||
headers = dict((k.decode('latin-1').lower(), v.decode('latin-1'))
|
||
for k, v in scope.get('headers', []))
|
||
auth = headers.get('authorization', '')
|
||
if API_KEY and auth != f'Bearer {API_KEY}':
|
||
resp = JSONResponse({'error': 'unauthorized: 缺少或错误的 Bearer API Key'}, status_code=401)
|
||
await resp(scope, receive, send)
|
||
return
|
||
await self.app(scope, receive, send)
|
||
|
||
|
||
def main():
|
||
app = mcp.http_app(path='/mcp/', stateless_http=True,
|
||
allowed_hosts=['api.rrrosmontis.icu', '127.0.0.1', 'localhost'])
|
||
app = BearerKeyMiddleware(app)
|
||
log.info(f'ros-mail-mcp listening on http://{HOST}:{PORT}/mcp/')
|
||
uvicorn.run(app, host=HOST, port=PORT, log_level='warning')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|