--- title: MCP Mail: permettre à OpenWebUI d'échanger avec une boite mail author: Frederic AOUSTIN version: 1.O last_date: 13/07/2026 --- # MCP Mail: permettre à OpenWebUI d'échanger avec une boite mail ![category](developpement) ![tag](python) ![tag](ia) Nous avons déjà réalisé quelques serveurs **MCP** et il ne partait pas difficile de créer un programme python permettant d'envoyer un mail. Mais la prise en compte de ce besoin soulève une nouvelle problématique: la gestion de l'identification au niveau du serveur mcp. En effet le standard **MCP** ne prevoit pas de gérer de façon native l'identification de l'utilisateur qui utilise le service. Ainsi il faut répondre à la question: comment savons nous qui demande depuis *OpenWebUi* de faire telle action. Pour répondre à cette problématique on peut activer la variable **ENABLE_FORWARD_USER_INFO_HEADERS=True** au niveau du serveur *OpenWebUI*. Une fois activé, Open WebUI enverra des headers comme : * X-OpenWebUI-User-Name * X-OpenWebUI-User-Id * X-OpenWebUI-User-Email * X-OpenWebUI-User-Role et selon la cible aussi : * X-OpenWebUI-Chat-Id * X-OpenWebUI-Message-Id Et ainsi on pourra utiliser ces variables pour récupérer une configuration spécifique lors de l'utilisation du service. Ceci n'est possible que si - le serveur mcp ne soit accessible uniquement à **OpenWebUi** (en effet sinon n'importe qui peut rentrer dans le header des demandes ces infos et *imiter* un autre utilisateur) - s'assurer que la création du compte dans **OpenWebUi** soit contrôle (pour éviter qu'unutilisateur génère par exemple un compte avec l'email d'un responsable) Par la suite il suffit de rajouter un middleware, en plus du middleware qui check la clé **Bearer**, gérant les clés dans le header. ## le serveur MCP le fichier *requirements.txt** ``` mcp==1.27.2 pydantic_core==2.46.4 requests==2.34.2 rich==15.0.0 ``` le fichier *Dockerfile* ``` FROM python:3.12-alpine WORKDIR /app RUN pip install --upgrade pip COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY main.py main.py ENV MCP_HOST=0.0.0.0 ENV MCP_PORT=8000 ENV MCP_LEVEL=METRIC ENV MCP_ENV=prod EXPOSE 8000 CMD ["python", "main.py"] ``` le fichier *main.py* ```python import os import re import hmac import datetime import logging from dotenv import load_dotenv from typing import Annotated from pydantic import BaseModel, Field from mcp.server.fastmcp import FastMCP from starlette.middleware.cors import CORSMiddleware from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import JSONResponse, StreamingResponse from starlette.concurrency import run_in_threadpool import uvicorn import requests ################ logging ########################################## import logging import functools import time from rich.theme import Theme import rich.themes def clean_log(root): for children in root.getChildren(): children.handlers.clear() children.propagate = True clean_log(children) def add_logging_level(level_name, level_num): logging.addLevelName(level_num, level_name) def log_for_level(self, message, *args, **kwargs): if self.isEnabledFor(level_num): self._log(level_num, message, args, **kwargs) def log_to_root(message, *args, **kwargs): logging.log(level_num, message, *args, **kwargs) setattr(logging.Logger, level_name.lower(), log_for_level) setattr(logging, level_name.lower(), log_to_root) add_logging_level("ANALYZE", 15) add_logging_level("AUDIT", 60) add_logging_level("METRIC", 16) add_logging_level("SECURITY", 80) def metric_root(arg=None, *args, **kwargs): if callable(arg): func = arg @functools.wraps(func) def wrapper(*f_args, **f_kwargs): logging.log(logging.getLevelName('INFO'), f"Call {func.__name__}") start = time.time() try: result = func(*f_args, **f_kwargs) except Exception as e: logging.log(logging.getLevelName('METRIC'), f"Completed {func.__name__} in {time.time() - start:.2f} seconds") logging.error(f"{func.__name__} : {e}") raise e logging.log(logging.getLevelName('METRIC'), f"Completed {func.__name__} in {time.time() - start:.2f} seconds") return result return wrapper else: message = arg return logging.log(logging.getLevelName('METRIC'), message, *args, **kwargs) logging.metric = metric_root rich.themes.DEFAULT = Theme({ "logging.level.audit": "purple", "logging.level.analyze": "grey66", "logging.level.metric": "dark_cyan", "logging.level.security": "magenta", "logging.level.error": "deep_pink2", "logging.level.critical": "bold red1" }) ####################################################################### __version__ = "1.0.0" load_dotenv(override=True) API_TOKEN = os.getenv('API_TOKEN', "fZpcLTAGfiFApoVODWez7Y7inwdHoG") MCP_HOST = os.getenv('MCP_HOST', '0.0.0.0') MCP_PORT = int(os.getenv('MCP_PORT', 8000)) LOG_LEVEL = os.getenv('MCP_LEVEL', 'METRIC') LOG_FORMAT = os.getenv('MCP_LOG_FORMAT', "%(asctime)s %(levelname)-8s %(message)s") DATE_FORMAT = os.getenv('MCP_DATE_FORMAT', "%Y/%m/%d %H:%M:%S") ENV = os.getenv('MCP_ENV', 'dev') OPENWEBUI_BASE_URL = os.getenv('OPENWEBUI_BASE_URL', 'http://127.0.0.1:8080').rstrip('/') API_TOKEN_OPENWEBUI = os.getenv('API_TOKEN_OPENWEBUI', '') TTL_HOURS = int(os.getenv('TTL_HOURS', '24')) PREFIX = os.getenv('PREFIX', 'genfile_') PUBLIC_BASE_URL = os.getenv('PUBLIC_BASE_URL', f'http://{MCP_HOST}:{MCP_PORT}').rstrip('/') if LOG_LEVEL not in logging.getLevelNamesMapping(): LOG_LEVEL = 'INFO' LOGGING_CONFIG = { "version": 1, "disable_existing_loggers": False, "formatters": { "default": { "format": LOG_FORMAT, "datefmt": DATE_FORMAT, }, }, "loggers": { "uvicorn": { "handlers": [], "level": LOG_LEVEL, "propagate": True }, "mcp": { "handlers": [], "level": LOG_LEVEL, "propagate": True, }, }, } mcp = FastMCP( "MCP Server File", host=MCP_HOST, port=MCP_PORT, ) def _headers() -> dict: """Builds the headers used to call the OpenWebUI API (includes auth).""" headers = {} if API_TOKEN_OPENWEBUI: headers["Authorization"] = f"Bearer {API_TOKEN_OPENWEBUI}" return headers def _safe_filename(filename: str) -> str: """ Sanitizes a caller-provided filename to prevent any path traversal ('../') or dangerous characters before using it in a request to the OpenWebUI API. """ filename = os.path.basename(filename.strip()) filename = re.sub(r"[^A-Za-z0-9._-]", "_", filename) if not filename: filename = "file.txt" return filename def _list_remote_files() -> list: """ Lists all files known to OpenWebUI, without their content. GET /api/v1/files/ is paginated (fixed page size server-side) and returns a wrapper object (an 'items' list) rather than a bare list on recent OpenWebUI versions. This walks every page and requests content=false so the (potentially large) file content is never downloaded just to get names/ids. Older OpenWebUI versions that still return a bare, non-paginated list are also supported. """ all_files = [] page = 1 max_pages = 200 # safety cap against an unexpected infinite loop while page <= max_pages: resp = requests.get( f"{OPENWEBUI_BASE_URL}/api/v1/files/", headers=_headers(), params={"content": "false", "page": page}, timeout=10, ) resp.raise_for_status() data = resp.json() if isinstance(data, dict): items = data.get("items") or data.get("data") or [] else: # Older / non-paginated OpenWebUI versions return a bare list. items = data if not items: break all_files.extend(items) # If the server ignored our pagination (bare list case), don't loop forever. if not isinstance(data, dict): break page += 1 return all_files def _cleanup_expired_files() -> int: """Internal cleanup logic, reused by several tools.""" try: files = _list_remote_files() except Exception as e: logging.error(f"cleanup: unable to list OpenWebUI files: {e}") return 0 threshold = datetime.datetime.now() - datetime.timedelta(hours=TTL_HOURS) deleted_count = 0 for f in files: name = f.get("filename", "") if not name.startswith(PREFIX): continue # never touch files uploaded manually by a user created_raw = f.get("created_at") if created_raw is None: continue try: created = datetime.datetime.fromtimestamp(created_raw) except (TypeError, ValueError, OSError): continue if created < threshold: try: requests.delete( f"{OPENWEBUI_BASE_URL}/api/v1/files/{f['id']}", headers=_headers(), timeout=10, ) deleted_count += 1 except Exception as e: logging.error(f"cleanup: failed to delete file {f.get('id')}: {e}") return deleted_count # --------------------------------------------------------------------- # Tools # --------------------------------------------------------------------- @mcp.tool( name="cleanup_expired_files", title="Cleanup Expired Files", ) @logging.metric def cleanup_expired_files() -> dict: """ Permanently deletes, on the OpenWebUI server, all files previously generated by the generate_file tool whose lifetime (TTL) has expired. Never touches files manually uploaded by a user (they don't carry the internal generated-file prefix). This tool is also called automatically every time a file is generated; calling it manually is mainly useful to free up disk space on demand or for debugging. """ deleted_count = _cleanup_expired_files() return { "status": "ok", "deleted_files": deleted_count, "ttl_hours": TTL_HOURS, } @mcp.tool( name="generate_file", title="Generate Downloadable File", ) @logging.metric def generate_file( content: Annotated[str, Field( description="The full text content of the file to generate (UTF-8 encoded)." )], filename: Annotated[str, Field( description=( "File name with its extension, e.g. 'report.txt' or " "'export.csv'. Non-alphanumeric characters are automatically " "replaced; no directory path is accepted." ) )], ) -> dict: """ Creates a text file from the given content and stores it on the OpenWebUI server, then returns a download link. The file is stored with an internal prefix and will be automatically deleted after the configured TTL. Previously expired files are cleaned up automatically before the new one is created. Use this tool for any content the user wants to download (report, export, code, logs, etc.) rather than just displaying it in the conversation. """ safe_name = _safe_filename(filename) deleted_count = _cleanup_expired_files() final_name = f"{PREFIX}{safe_name}" files = {"file": (final_name, content.encode("utf-8"))} try: resp = requests.post( f"{OPENWEBUI_BASE_URL}/api/v1/files/", files=files, headers=_headers(), timeout=15, ) resp.raise_for_status() file_id = resp.json()["id"] except Exception as e: logging.error(f"generate_file: upload to OpenWebUI failed: {e}") return { "status": "error", "message": f"Could not create the file on the OpenWebUI server: {e}", } download_url = f"{PUBLIC_BASE_URL}/download/{file_id}" return { "status": "ok", "filename": safe_name, "file_id": file_id, "download_url": download_url, "markdown_link": f"[📄 Download {safe_name}]({download_url})", "ttl_hours": TTL_HOURS, "expired_files_cleaned": deleted_count, } @mcp.tool( name="delete_file_now", title="Delete Generated File Immediately", ) @logging.metric def delete_file_now( filename: Annotated[str, Field( description="File name as given to the user (without the internal prefix)." )], ) -> dict: """ Immediately deletes a file previously created with generate_file, without waiting for the TTL to expire. The provided name must be the one given to the user (the internal prefix is added automatically). Cannot be used to delete a file that wasn't created by this tool. """ safe_name = _safe_filename(filename) search_name = f"{PREFIX}{safe_name}" try: files = _list_remote_files() except Exception as e: logging.error(f"delete_file_now: unable to list files: {e}") return {"status": "error", "message": f"Could not reach the OpenWebUI server: {e}"} for f in files: if f.get("filename") == search_name: try: requests.delete( f"{OPENWEBUI_BASE_URL}/api/v1/files/{f['id']}", headers=_headers(), timeout=10, ) except Exception as e: logging.error(f"delete_file_now: deletion failed: {e}") return {"status": "error", "message": f"Deletion failed: {e}"} return {"status": "ok", "filename": safe_name, "message": f"File {safe_name} deleted."} return {"status": "not_found", "filename": safe_name, "message": f"No file named {safe_name} found."} async def download_endpoint(request: Request): """ Public, unauthenticated download proxy. Fetches the file from OpenWebUI using this server's own privileged credentials (_headers()) and streams it back to whoever has the link. This exists because OpenWebUI enforces per-owner file access control on its native /api/v1/files/{id}/content endpoint: only the uploading user (our technical account) or an admin can download it there. Proxying through here makes the link work for every OpenWebUI user (and, in fact, anyone who has the URL — treat it like a share link). """ file_id = request.path_params["file_id"] def _fetch_metadata(): r = requests.get( f"{OPENWEBUI_BASE_URL}/api/v1/files/{file_id}", headers=_headers(), params={"content": "false"}, timeout=10, ) return r meta_resp = await run_in_threadpool(_fetch_metadata) if meta_resp.status_code == 404: return JSONResponse({"error": "File not found"}, status_code=404) meta_resp.raise_for_status() meta = meta_resp.json() stored_name = meta.get("filename", file_id) display_name = stored_name[len(PREFIX):] if stored_name.startswith(PREFIX) else stored_name content_type = (meta.get("meta") or {}).get("content_type") or "application/octet-stream" def _fetch_content(): return requests.get( f"{OPENWEBUI_BASE_URL}/api/v1/files/{file_id}/content", headers=_headers(), params={"attachment": "true"}, timeout=30, stream=True, ) content_resp = await run_in_threadpool(_fetch_content) if content_resp.status_code == 404: return JSONResponse({"error": "File not found"}, status_code=404) content_resp.raise_for_status() return StreamingResponse( content_resp.iter_content(chunk_size=65536), media_type=content_type, headers={"Content-Disposition": f'attachment; filename="{display_name}"'}, ) async def version_endpoint(request: Request): return JSONResponse({ "name": "Files Server", "version": __version__, "protocol": "2026-07-02", "env": ENV, "log": LOG_LEVEL }) class TokenAuthMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): if request.url.path in ("/version", "/favicon.ico") or request.url.path.startswith("/download/"): return await call_next(request) if request.method == "OPTIONS": return await call_next(request) body = {} call = request.method try: body = await request.json() logging.analyze(body) call = body.get('method', 'unknown') if 'params' in body and 'name' in body.get('params'): call = body.get('params').get('name', 'unknown') except Exception: pass auth = request.headers.get("Authorization", "") if auth != f"Bearer {API_TOKEN}": logging.security(f"Error Authentification {call} {auth}") return JSONResponse( {"error": "Unauthorized"}, status_code=401 ) logging.security(f"Authentification valided {call}") return await call_next(request) if __name__ == "__main__": app = mcp.streamable_http_app() from starlette.routing import Route app.routes.append(Route("/version", version_endpoint)) app.routes.append(Route("/download/{file_id}", download_endpoint)) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], expose_headers=["mcp-session-id"], allow_credentials=False, ) app.add_middleware(TokenAuthMiddleware) root = logging.getLogger() root.handlers.clear() clean_log(root) if ENV == 'dev': from rich.logging import RichHandler root.addHandler(RichHandler( show_time=True, show_level=True, show_path=True, log_time_format=DATE_FORMAT, omit_repeated_times=False,)) else: formatter = logging.Formatter(fmt=LOG_FORMAT, datefmt=DATE_FORMAT) handler = logging.StreamHandler() handler.setFormatter(formatter) root.addHandler(handler) root.setLevel(logging.getLevelNamesMapping().get(LOG_LEVEL, logging.INFO)) logging.info(f"Starting File v{__version__}") logging.security(f"Authentification by token {API_TOKEN}") uvicorn.run(app, host=MCP_HOST, port=MCP_PORT, log_config=LOGGING_CONFIG) ``` Au niveau du *docker-compose.yml* ``` mcpmail: build: sam/mcp-mail/. container_name: mcpmail environment: - API_TOKEN=YYYYYYYYYYYYYYYYYYYYYY - MCP_HOST=0.0.0.0 - MCP_PORT=8000 - MCP_LEVEL=METRIC - MCP_ENV=prod - mail_mail_openwebui_fr=smtps://user%40gmail.com:password_mail@smtp.gmail.com:465 restart: always networks: meet.jitsi: ``` Il faut noter qu'il faut ajouter une variable par utilisateur composé ainsi - mail_mail_sanitize: il faut remplacer les @ et . du mail d'openwebui par un _ - format de la valeur smtp://user:password@host:port ... si vous utilisez du ssl il faudra indiquer *smtps* (/!\ au caractère spéciaux dans le mail et password)