MCP server source
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
FROM python:3.13-slim
|
||||
WORKDIR /app
|
||||
COPY requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
COPY server.py ./
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
EXPOSE 8902
|
||||
CMD ["python3", "server.py"]
|
||||
@@ -0,0 +1 @@
|
||||
fastmcp>=3.0.0
|
||||
@@ -0,0 +1,68 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Email MCP Server — Send and check email from Open WebUI."""
|
||||
from fastmcp import FastMCP
|
||||
import smtplib, imaplib, email, os
|
||||
from email.mime.text import MIMEText
|
||||
|
||||
mcp = FastMCP('Email', version='1.0.0')
|
||||
|
||||
SMTP_HOST = 'mail.germainebrown.com'
|
||||
SMTP_PORT = 2525
|
||||
IMAP_HOST = 'mail.germainebrown.com'
|
||||
IMAP_PORT = 993
|
||||
|
||||
def _get_pass(filename):
|
||||
with open(f'/root/.config/himalaya/{filename}') as f:
|
||||
return f.read().strip()
|
||||
|
||||
@mcp.tool(description="Send an email from g@germainebrown.com")
|
||||
async def send_email(to: str, subject: str, body: str) -> str:
|
||||
"""Send an email via SMTP."""
|
||||
pw = _get_pass('g-germainebrown.pass')
|
||||
msg = MIMEText(body)
|
||||
msg['From'] = 'g@germainebrown.com'
|
||||
msg['To'] = to
|
||||
msg['Subject'] = subject
|
||||
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as s:
|
||||
s.starttls()
|
||||
s.login('g@germainebrown.com', pw)
|
||||
s.send_message(msg)
|
||||
return f'Sent email to {to}: "{subject}"'
|
||||
|
||||
@mcp.tool(description="Check recent emails in shonuff@germainebrown.com inbox")
|
||||
async def check_inbox(limit: int = 10) -> str:
|
||||
"""Read recent inbox emails."""
|
||||
pw = _get_pass('shonuff.pass')
|
||||
with imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT) as imap:
|
||||
imap.login('shonuff@germainebrown.com', pw)
|
||||
imap.select('INBOX')
|
||||
_, msgs = imap.search(None, 'ALL')
|
||||
ids = msgs[0].split()[-limit:]
|
||||
results = []
|
||||
for mid in reversed(ids):
|
||||
_, data = imap.fetch(mid, '(RFC822)')
|
||||
msg = email.message_from_bytes(data[0][1])
|
||||
results.append(f'From: {msg["From"]} | Subject: {msg["Subject"]} | Date: {msg["Date"]}')
|
||||
return f'Last {len(results)} emails:\n' + '\n'.join(results)
|
||||
|
||||
@mcp.tool(description="Search shonuff@ inbox by subject or sender")
|
||||
async def search_email(query: str, limit: int = 10) -> str:
|
||||
"""Search inbox."""
|
||||
pw = _get_pass('shonuff.pass')
|
||||
with imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT) as imap:
|
||||
imap.login('shonuff@germainebrown.com', pw)
|
||||
imap.select('INBOX')
|
||||
_, msgs = imap.search(None, f'SUBJECT "{query}"')
|
||||
subject_ids = msgs[0].split()
|
||||
_, msgs = imap.search(None, f'FROM "{query}"')
|
||||
from_ids = msgs[0].split()
|
||||
all_ids = list(set(list(subject_ids) + list(from_ids)))[-limit:]
|
||||
results = []
|
||||
for mid in reversed(all_ids):
|
||||
_, data = imap.fetch(mid, '(RFC822)')
|
||||
msg = email.message_from_bytes(data[0][1])
|
||||
results.append(f'From: {msg["From"]} | Subject: {msg["Subject"]} | Date: {msg["Date"]}')
|
||||
return f'Found {len(results)} matches for "{query}":\n' + '\n'.join(results)
|
||||
|
||||
if __name__ == '__main__':
|
||||
mcp.run(transport='http', host='0.0.0.0', port=8902, path='/mcp')
|
||||
Reference in New Issue
Block a user