--- title: MCP File: permettre à OpenWebUI d'échanger des fichiers author: Frederic AOUSTIN version: 1.O last_date: 05/07/2026 --- # MCP File: permettre à OpenWebUI d'échanger des fichiers ![category](developpement) ![tag](python) ![tag](ia) Ce que je demande souvent à un LLM est quelques choses de type: "Génère moi un fichier comprenant ..." ... si par défaut le LLM sait me répondre il n'a pas d'outil permettant de me partager ce fichier. Le plus souvent il génère un bout de texte qu'il me demande de copier/coller. Il existe déjà des serveurs mcp qui gère les fichiers ... mais cela à souvent pour objectif de gérer des fichiers locaux dans le cas d'usage d'un orchestrateur local ... pas d'un usage à travers le client web. Si on installe de tels serveurs, le LLM pourra créer localement sur le serveur un fichier mais il faudra par la suite donner accès à l'utilisateur à ce fichier ... ce qui n'est pas évident et doit être fait par une autre brique. J'ai donc créer un serveur **mcp** qui réutilise la capacité de **OpenWebUI** à gérer du fichier. En faite on utilise un utilisateur technique qui va permettre d'enregistrer dans sa base de connaissance le fichier généré puis partager l'url dans la réponse du LLM. ## Création du compte technique Il faut créer un compte **technique** qui devra être dans un groupe **technique** avec le droit de faire des apis. ![img](./upload/20260708115700.png) Il faut par la suite il faut se connecter avec cet utilisateur et créer une clé API qu'on pourra utiliser au niveau serveur mcp ![img](./upload/20260708115701.png) ## Gestion du serveur MCP Comme d'habitude on le génère dans le panneau d'administration / Reglages / **Intégrations** Créer le serveur avec le mode **Streamable HTTP** et fournir un id Il faut aussi gérer les accès pour le donner à l'ensemble des autres utilisateurs. Vous Pouvez aussi ajouter cet outil aux LLM que vous avez créer afin qu'il soit pris en charge par défaut par ces LLMs. ## Le code On va gérer ce code sous format *Docker* Mon *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"] ``` Mon fichier *requirements.txt* ``` mcp==1.27.2 pydantic_core==2.46.4 requests==2.34.2 rich==15.0.0 ``` Mon 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) ``` Exemple d'intégration dans un *docker-compose.yml* ``` open-webui: image: ghcr.io/open-webui/open-webui:0.10.2 #image: ghcr.io/open-webui/open-webui:0.9.6 container_name: open-webui volumes: - ./data/sam/open-webui:/app/backend/data depends_on: - proxyollama ports: - 8088:8080 labels: - "traefik.enable=true" - "traefik.http.routers.ia.entrypoints=websecure" - "traefik.http.routers.ia.tls=true" - "traefik.http.routers.ia.rule=Host(`ia.fraoustin.fr`) || Host(`ai.fraoustin.fr`)" - "traefik.http.routers.ia.priority=1" - "traefik.http.services.ia-service.loadbalancer.server.port=8080" - "traefik.http.routers.ia-insecure.entrypoints=web" - "traefik.http.routers.ia-insecure.rule=Host(`ia.fraoustin.fr`) || Host(`ai.fraoustin.fr`)" - "traefik.http.routers.ia-insecure.priority=1" - "traefik.http.routers.ia-insecure.middlewares=https-redirect" environment: - OLLAMA_BASE_URL=http://proxyollama:11434 - WEBUI_SECRET_KEY=6f0d8f7b2c5e1a4d9f3e7c8b1a6d5e2f9c4b7a1d8e6f3c5b extra_hosts: - host.docker.internal:host-gateway restart: always networks: meet.jitsi: mcpfile: build: sam/mcp-file/. container_name: mcpfile environment: - API_TOKEN=fZpcLTAGfiFApoVODWez7Y7inwdHoG - MCP_HOST=0.0.0.0 - MCP_PORT=8000 - MCP_LEVEL=METRIC - MCP_ENV=prod - OPENWEBUI_BASE_URL=https://ia.fraoustin.fr - API_TOKEN_OPENWEBUI=sk-d7ae48199d144506b7a4514d41a135cf - PUBLIC_BASE_URL=https://ia.fraoustin.fr restart: always labels: - "traefik.enable=true" - "traefik.http.routers.mcpfiles.entrypoints=websecure" - "traefik.http.routers.mcpfiles.tls=true" - "traefik.http.routers.mcpfiles.rule=Host(`ia.fraoustin.fr`) && PathPrefix(`/download`)" - "traefik.http.routers.mcpfiles.priority=100" - "traefik.http.services.mcpfiles-service.loadbalancer.server.port=8000" - "traefik.http.routers.mcpfiles-insecure.entrypoints=web" - "traefik.http.routers.mcpfiles-insecure.rule=Host(`ia.fraoustin.fr`) && PathPrefix(`/download`)" - "traefik.http.routers.mcpfiles-insecure.priority=100" - "traefik.http.routers.mcpfiles-insecure.middlewares=https-redirect" networks: meet.jitsi: ```