MCP tools usually return text. But a tool can also return media (an image, a document), and sometimes the right answer isn’t the bytes at all but a link to them. This tutorial builds a FastMCP server that does both, and lets you choose per request:
- Inline delivery — the tool returns the media object directly (a PNG image, an SVG image, a Markdown document). Best for small, renderable results.
- Link delivery — the tool stores the bytes in a bucket and returns a URL. The URL is either permanent or expiring after N minutes. Best for large media (video) or anything you want to share by link.
The bucket here is self-contained: the server stores files in a local directory
and serves them itself over HTTP, using HMAC-signed URLs so an expiring link
simply stops working after its time-to-live — no per-object cleanup job. The
last section shows how to swap in real object storage (S3/R2/GCS presigned URLs)
without touching the tools. The stack is Mac-native: uv, make, and a
launchd service.
What you will build
- Inline tools:
swatch_png(PNG image),badge_svg(SVG image),note_markdown(Markdown text). - Link tools:
badge_svg_link(store a generated badge) andstore_media(store arbitrary bytes, e.g. video), each returning a permanent or expiring URL. - A
/media/<key>HTTP route on the same server that serves stored files and enforces expiry and signatures. - A
pytestsuite covering the bucket logic, the tool outputs, and the HTTP route end to end.
Prerequisites
- macOS 13+ with Homebrew (brew.sh).
- uv 0.5+ —
brew install uv; verifyuv --version. - Xcode Command Line Tools (
xcode-select --install) formake. - Familiarity with Python type hints and a passing acquaintance with HMAC (used to sign links; explained where it appears).
Everything runs through uv and make. Media serving requires the HTTP
transport (MCP_TRANSPORT=http); stdio has no HTTP endpoint for links.
Step 1: Scaffold and lock down hygiene
Create the workspace and a .gitignore first. Note it ignores media/ — the
bucket contents are runtime data, not source.
Create the files
mkdir -p media-fastmcp-server-macos
cd media-fastmcp-server-macos
touch .gitignore
Add the code: .gitignore
# Python
__pycache__/
*.py[cod]
.venv/
.uv/
.pytest_cache/
.ruff_cache/
.mypy_cache/
# Runtime / deploy artifacts
logs/
*.log
*.pid
media/
# OS / editor noise
.DS_Store
Detailed breakdown
media/is the bucket directory the server writes to at runtime. It holds generated/uploaded files, never source, so it stays out of git.logs/holds the deployed service’s output; the rest is standarduv/macOS hygiene.
Step 2: Initialize the project with uv
Create the uv project and add FastMCP plus the test tools.
Create the files
uv init --name macmcp --no-workspace
rm -f main.py hello.py
uv add fastmcp
uv add --dev pytest pytest-asyncio
Add the code: pyproject.toml
[project]
name = "macmcp"
version = "0.1.0"
description = "A FastMCP media server with inline and bucket-link delivery"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"fastmcp>=3.4.3",
]
[dependency-groups]
dev = [
"pytest>=9.1.1",
"pytest-asyncio>=1.4.0",
]
Detailed breakdown
fastmcpbrings the server, theImage/Filemedia helpers, andstarlette(a transitive dependency) which we use for the media route and its tests — no extra dependencies needed.
Step 3: Write the media generators
Keep media generation in a dependency-free module of pure functions. Each returns raw bytes (or text), which makes them trivial to test and reuse for both delivery strategies. The PNG is hand-built with the standard library — no Pillow.
Create the file
touch media.py
Add the code: media.py
"""Pure media generators — deterministic bytes, no third-party dependencies."""
import base64
import struct
import zlib
def svg_badge(label: str, message: str) -> bytes:
"""Render a simple two-tone SVG badge as bytes."""
lw, mw = 6 * len(label) + 10, 6 * len(message) + 10
total = lw + mw
svg = f"""<svg xmlns="http://www.w3.org/2000/svg" width="{total}" height="20">
<rect width="{lw}" height="20" fill="#555"/>
<rect x="{lw}" width="{mw}" height="20" fill="#4c1"/>
<g fill="#fff" font-family="Verdana" font-size="11">
<text x="5" y="14">{label}</text>
<text x="{lw + 5}" y="14">{message}</text>
</g>
</svg>"""
return svg.encode()
def png_swatch(hex_color: str, size: int = 32) -> bytes:
"""Build a solid-color PNG of `size`x`size` pixels, using only the stdlib."""
hex_color = hex_color.lstrip("#")
if len(hex_color) == 3: # expand shorthand like "4c1" -> "44cc11"
hex_color = "".join(c * 2 for c in hex_color)
if len(hex_color) != 6:
raise ValueError(f"expected a 3- or 6-digit hex color, got {hex_color!r}")
r, g, b = (int(hex_color[i : i + 2], 16) for i in (0, 2, 4))
raw = bytearray()
for _ in range(size):
raw.append(0) # filter byte per scanline
raw.extend((r, g, b) * size)
def chunk(tag: bytes, data: bytes) -> bytes:
body = tag + data
return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body))
ihdr = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0) # 8-bit RGB
return (
b"\x89PNG\r\n\x1a\n"
+ chunk(b"IHDR", ihdr)
+ chunk(b"IDAT", zlib.compress(bytes(raw)))
+ chunk(b"IEND", b"")
)
def markdown_note(title: str, body: str) -> str:
"""Render a small Markdown document as text."""
return f"# {title}\n\n{body}\n"
def decode_upload(content_base64: str) -> bytes:
"""Decode a base64 payload a client sent for storage."""
return base64.b64decode(content_base64)
Detailed breakdown
svg_badgereturns UTF-8 SVG bytes. SVG is text, so this is just string formatting — but as media it carries theimage/svg+xmltype once served.png_swatchassembles a valid PNG by hand: the 8-byte signature, anIHDRheader (dimensions, 8-bit RGB), the pixel data zlib-compressed intoIDAT(each scanline prefixed with a filter byte), and theIENDterminator, each chunk CRC-checked. This keeps the project dependency-free and demonstrates returning true binary media. Shorthand hex like#4c1is expanded first.decode_uploadturns a base64 string (how a client hands you binary over JSON) back into bytes for storage — the entry point for arbitrary media such as video.
Step 4: Write the bucket — permanent and expiring links
The bucket stores bytes and mints links. A permanent link is just the
object’s path. An expiring link appends an exp timestamp and an HMAC sig;
the server recomputes the signature and checks the clock. Because expiry is
encoded in the signed URL, nothing has to delete files on a schedule — an expired
link is simply refused.
Create the file
touch bucket.py
Add the code: bucket.py
"""A tiny self-contained media bucket with permanent and expiring links.
Files are written to a local directory (the "bucket") and served by the MCP
server itself over HTTP. A permanent link is just the object path; an expiring
link carries an `exp` timestamp and an HMAC `sig` that the server verifies, so a
link stops working after its time-to-live without any per-object cleanup.
In production, swap this module for real object storage: `store` becomes an
upload, `permanent_url` returns the object's public URL, and `signed_url`
becomes a presigned GET URL (S3/R2 `generate_presigned_url`, GCS signed URLs) —
which already carry native expiry. The tool code and the server stay the same.
"""
import hashlib
import hmac
import mimetypes
import os
import re
import secrets
import time
from pathlib import Path
DEFAULT_SECRET = "dev-insecure-change-me" # override with MEDIA_SECRET in prod
def _bucket_dir() -> Path:
path = Path(os.environ.get("MEDIA_DIR", "media"))
path.mkdir(parents=True, exist_ok=True)
return path
def _secret() -> bytes:
return os.environ.get("MEDIA_SECRET", DEFAULT_SECRET).encode()
def _base_url() -> str:
return os.environ.get("MEDIA_BASE_URL", "http://127.0.0.1:8000").rstrip("/")
def _safe(name: str) -> str:
"""Reduce a filename to a safe basename (no paths, no surprises)."""
name = os.path.basename(name)
return re.sub(r"[^A-Za-z0-9._-]", "_", name) or "file"
def store(data: bytes, filename: str) -> str:
"""Write bytes into the bucket and return the object's key."""
key = f"{secrets.token_hex(8)}-{_safe(filename)}"
(_bucket_dir() / key).write_bytes(data)
return key
def _sign(key: str, exp: int) -> str:
return hmac.new(_secret(), f"{key}:{exp}".encode(), hashlib.sha256).hexdigest()
def permanent_url(key: str) -> str:
return f"{_base_url()}/media/{key}"
def signed_url(key: str, ttl_minutes: int) -> str:
exp = int(time.time()) + ttl_minutes * 60
return f"{_base_url()}/media/{key}?exp={exp}&sig={_sign(key, exp)}"
def verify(key: str, exp: str | None, sig: str | None) -> tuple[bool, str]:
"""Authorize a request for `key`. No exp/sig means a permanent link."""
if exp is None and sig is None:
return True, "ok"
if exp is None or sig is None:
return False, "malformed link"
try:
exp_int = int(exp)
except ValueError:
return False, "malformed link"
if not hmac.compare_digest(sig, _sign(key, exp_int)):
return False, "bad signature"
if time.time() > exp_int:
return False, "link expired"
return True, "ok"
def read(key: str) -> bytes | None:
"""Return the object's bytes, or None. Refuses to escape the bucket."""
root = _bucket_dir().resolve()
target = (root / _safe(key)).resolve()
if target.parent != root or not target.is_file():
return None
return target.read_bytes()
def content_type(key: str) -> str:
return mimetypes.guess_type(key)[0] or "application/octet-stream"
Detailed breakdown
storeprefixes a random token to a sanitized filename, so keys are unguessable and collision-free, and writes the bytes. It returns the key used in URLs._sign/signed_urlproduce the expiring link: the signature isHMAC(secret, "key:exp"). Only someone holdingMEDIA_SECRETcan mint a valid link, and theexpis inside the signed material so it cannot be extended by editing the URL.verifyis the gate the HTTP route calls. Noexp/sig→ a permanent link, allowed. Otherwise it useshmac.compare_digest(constant-time, to avoid timing attacks) and then checks expiry. It returns a reason string for a clear 403.readrefuses path traversal: it resolves the target and confirms its parent is exactly the bucket root, so a crafted key like../server.pyreturnsNone.content_typegives the served file the correct MIME type viamimetypes.- The module docstring is the porting guide: every function has a one-to-one mapping onto S3/R2/GCS, so moving to real storage never touches the tools.
Step 5: Write the server — tools and the media route
Now the server. Inline tools return Image objects or text; link tools call the
bucket and return a URL. A single @mcp.custom_route serves the bucket over the
same HTTP port the MCP endpoint uses.
Create the file
touch server.py
Add the code: server.py
"""A FastMCP server that returns media inline or as bucket links.
Two delivery strategies:
- **Inline** — the tool returns the media object directly in its result
(an image or text content block). Best for small, renderable media.
- **Link** — the tool stores the bytes in a bucket and returns a URL, either
permanent or expiring after N minutes. Best for large media (video) or when
you want a shareable link. The same server serves the bucket over HTTP.
Auth/media serving works over the HTTP transport; run with MCP_TRANSPORT=http.
"""
import os
from fastmcp import FastMCP
from fastmcp.utilities.types import Image
from starlette.requests import Request
from starlette.responses import Response
import bucket
import media
mcp = FastMCP("macmcp")
# --- Inline delivery: return the media object directly ------------------------
@mcp.tool
def swatch_png(color: str, size: int = 32) -> Image:
"""Return a solid-color PNG swatch inline (color is a hex like #4c1)."""
return Image(data=media.png_swatch(color, size), format="png")
@mcp.tool
def badge_svg(label: str, message: str) -> Image:
"""Return an SVG badge inline."""
return Image(data=media.svg_badge(label, message), format="svg")
@mcp.tool
def note_markdown(title: str, body: str) -> str:
"""Return a Markdown document inline (as text)."""
return media.markdown_note(title, body)
# --- Link delivery: store in the bucket, return a URL -------------------------
def _link(key: str, ttl_minutes: int) -> dict:
if ttl_minutes <= 0:
return {"url": bucket.permanent_url(key), "expires": "never"}
return {
"url": bucket.signed_url(key, ttl_minutes),
"expires": f"in {ttl_minutes} minute(s)",
}
@mcp.tool
def badge_svg_link(label: str, message: str, ttl_minutes: int = 0) -> dict:
"""Store an SVG badge and return a link. ttl_minutes=0 means permanent."""
key = bucket.store(media.svg_badge(label, message), "badge.svg")
return _link(key, ttl_minutes)
@mcp.tool
def store_media(filename: str, content_base64: str, ttl_minutes: int = 0) -> dict:
"""Store arbitrary media bytes (png, mp4, ...) and return a link.
Use this for large media such as video, or any bytes the client already has.
ttl_minutes=0 returns a permanent link; a positive value expires the link.
"""
key = bucket.store(media.decode_upload(content_base64), filename)
return _link(key, ttl_minutes)
# --- The bucket: serve stored media over HTTP ---------------------------------
@mcp.custom_route("/media/{key}", methods=["GET"])
async def serve_media(request: Request) -> Response:
key = request.path_params["key"]
ok, reason = bucket.verify(
key, request.query_params.get("exp"), request.query_params.get("sig")
)
if not ok:
return Response(reason, status_code=403)
data = bucket.read(key)
if data is None:
return Response("not found", status_code=404)
return Response(data, media_type=bucket.content_type(key))
def main() -> None:
if os.environ.get("MCP_TRANSPORT", "stdio").lower() == "http":
mcp.run(
transport="http",
host=os.environ.get("MCP_HOST", "127.0.0.1"),
port=int(os.environ.get("MCP_PORT", "8000")),
)
else:
mcp.run() # stdio (media links still resolve only when HTTP is running)
if __name__ == "__main__":
main()
Detailed breakdown
- Inline tools return media objects.
Image(data=..., format="png")produces an MCP image content block (image/png); the SVG variant produces animage/svgblock.note_markdownreturns a plainstr, which becomes a text block. An MCP client renders these directly — no fetch required. - Link tools return a small dict with the
urland a human-readableexpires._linkchooses permanent vs signed based onttl_minutes.store_mediais the general path for video and any client-supplied bytes: the client base64-encodes the media, and the tool stores it and returns a link. @mcp.custom_route("/media/{key}")adds an ordinary HTTP GET endpoint to the same Starlette app FastMCP serves. It callsbucket.verify(permanent vs signed/expiring), returns403with the reason on failure,404if the object is gone, and otherwise streams the bytes with the rightContent-Type. This is why links minted by the tools actually resolve.- A note on when to use which: inline is best for small, renderable media a model or UI will show immediately; links are best for large media (inline base64 bloats the token stream) or when the media outlives the tool call.
Step 6: Test the bucket, the tools, and the route
Three test files, all hermetic. Bucket logic and tool outputs run in-memory;
the HTTP route runs through Starlette’s TestClient against the FastMCP app, so
you get real 200/403/404 responses with no live server.
Create the files
mkdir -p tests
touch tests/__init__.py tests/test_bucket.py tests/test_tools.py tests/test_media_route.py pytest.ini
Add the code: pytest.ini
[pytest]
asyncio_mode = auto
filterwarnings =
ignore::DeprecationWarning
Add the code: tests/test_bucket.py
"""Test the bucket's storage and permanent/expiring link logic."""
import time
import pytest
import bucket
@pytest.fixture(autouse=True)
def isolated_bucket(tmp_path, monkeypatch):
monkeypatch.setenv("MEDIA_DIR", str(tmp_path / "media"))
monkeypatch.setenv("MEDIA_SECRET", "test-secret")
monkeypatch.setenv("MEDIA_BASE_URL", "http://127.0.0.1:8000")
def test_store_and_read_roundtrip():
key = bucket.store(b"hello", "greeting.txt")
assert bucket.read(key) == b"hello"
def test_permanent_link_has_no_signature():
key = bucket.store(b"x", "a.png")
url = bucket.permanent_url(key)
assert url.endswith(f"/media/{key}")
assert "sig=" not in url
def test_permanent_link_always_verifies():
key = bucket.store(b"x", "a.png")
ok, _ = bucket.verify(key, None, None)
assert ok
def test_signed_link_verifies_before_expiry():
key = bucket.store(b"x", "a.png")
url = bucket.signed_url(key, ttl_minutes=10)
exp = url.split("exp=")[1].split("&")[0]
sig = url.split("sig=")[1]
ok, reason = bucket.verify(key, exp, sig)
assert ok, reason
def test_expired_link_is_rejected():
key = bucket.store(b"x", "a.png")
past = str(int(time.time()) - 5)
sig = bucket._sign(key, int(past))
ok, reason = bucket.verify(key, past, sig)
assert not ok and reason == "link expired"
def test_tampered_signature_is_rejected():
key = bucket.store(b"x", "a.png")
url = bucket.signed_url(key, ttl_minutes=10)
exp = url.split("exp=")[1].split("&")[0]
ok, reason = bucket.verify(key, exp, "deadbeef")
assert not ok and reason == "bad signature"
def test_read_refuses_path_traversal():
assert bucket.read("../server.py") is None
Add the code: tests/test_tools.py
"""Test the tools in-memory: inline media content and link shapes."""
import base64
import pytest
from fastmcp import Client
from server import mcp
@pytest.fixture(autouse=True)
def isolated_bucket(tmp_path, monkeypatch):
monkeypatch.setenv("MEDIA_DIR", str(tmp_path / "media"))
monkeypatch.setenv("MEDIA_SECRET", "test-secret")
monkeypatch.setenv("MEDIA_BASE_URL", "http://127.0.0.1:8000")
async def test_swatch_png_returns_inline_image():
async with Client(mcp) as client:
result = await client.call_tool("swatch_png", {"color": "#4c1"})
block = result.content[0]
assert block.type == "image"
assert block.mimeType == "image/png"
async def test_note_markdown_returns_text():
async with Client(mcp) as client:
result = await client.call_tool(
"note_markdown", {"title": "Hi", "body": "Body text"}
)
assert result.data == "# Hi\n\nBody text\n"
async def test_badge_link_permanent_by_default():
async with Client(mcp) as client:
result = await client.call_tool(
"badge_svg_link", {"label": "build", "message": "passing"}
)
assert result.data["expires"] == "never"
assert "sig=" not in result.data["url"]
async def test_badge_link_expiring():
async with Client(mcp) as client:
result = await client.call_tool(
"badge_svg_link",
{"label": "build", "message": "passing", "ttl_minutes": 5},
)
assert result.data["expires"] == "in 5 minute(s)"
assert "exp=" in result.data["url"] and "sig=" in result.data["url"]
async def test_store_media_accepts_arbitrary_bytes():
payload = base64.b64encode(b"\x00\x00fake mp4 bytes").decode()
async with Client(mcp) as client:
result = await client.call_tool(
"store_media",
{"filename": "clip.mp4", "content_base64": payload, "ttl_minutes": 1},
)
assert "clip.mp4" in result.data["url"]
assert "exp=" in result.data["url"]
Add the code: tests/test_media_route.py
"""Test the /media HTTP route end to end with Starlette's TestClient."""
import pytest
from starlette.testclient import TestClient
import bucket
from server import mcp
@pytest.fixture()
def client(tmp_path, monkeypatch):
monkeypatch.setenv("MEDIA_DIR", str(tmp_path / "media"))
monkeypatch.setenv("MEDIA_SECRET", "test-secret")
monkeypatch.setenv("MEDIA_BASE_URL", "http://testserver")
with TestClient(mcp.http_app()) as c:
yield c
def _path(url: str) -> str:
return url.replace("http://testserver", "")
def test_permanent_link_serves_200(client):
key = bucket.store(b"<svg/>", "badge.svg")
resp = client.get(_path(bucket.permanent_url(key)))
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("image/svg")
def test_valid_signed_link_serves_200(client):
key = bucket.store(b"data", "a.png")
resp = client.get(_path(bucket.signed_url(key, ttl_minutes=10)))
assert resp.status_code == 200
def test_tampered_signature_403(client):
key = bucket.store(b"data", "a.png")
url = _path(bucket.signed_url(key, ttl_minutes=10))
tampered = url.rsplit("sig=", 1)[0] + "sig=deadbeef"
assert client.get(tampered).status_code == 403
def test_expired_link_403(client):
key = bucket.store(b"data", "a.png")
exp = 1 # 1970 — long past
sig = bucket._sign(key, exp)
assert client.get(f"/media/{key}?exp={exp}&sig={sig}").status_code == 403
def test_missing_object_404(client):
resp = client.get("/media/does-not-exist-file.png")
assert resp.status_code == 404
Detailed breakdown
isolated_bucketpointsMEDIA_DIRat pytest’stmp_pathand pins the secret, so tests never touch a real bucket and signatures are reproducible.test_bucket.pycovers the security-critical logic directly: round-trip, permanent-vs-signed shape, valid-before-expiry, rejection of expired and tampered links, and path-traversal refusal.test_tools.pyasserts the inline image content type and text, and the permanent/expiring link shapes — the tool contract clients depend on.test_media_route.pyexercises the real HTTP route viamcp.http_app()+TestClient:200for permanent and valid-signed,403for tampered and expired,404for missing — the full server behavior with no network.
Run them:
uv run pytest -v
You should see all tests pass.
Step 7: The Makefile
Wrap development and the launchd deployment in make. Plain make prints
help.
Create the file
touch Makefile
Add the code: Makefile
.DEFAULT_GOAL := help
# --- Deploy configuration (launchd user agent) --------------------------------
LABEL := com.macmcp.media
PLIST := $(HOME)/Library/LaunchAgents/$(LABEL).plist
UV := $(shell command -v uv)
DIR := $(shell pwd)
DOMAIN := gui/$(shell id -u)
.PHONY: help install serve test demo deploy undeploy status logs clean
help: ## Show this help screen
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
| awk 'BEGIN {FS = ":.*?## "}; {printf " \033[36m%-10s\033[0m %s\n", $$1, $$2}'
install: ## Sync runtime and dev dependencies
uv sync
serve: ## Run the HTTP server in the foreground (Ctrl-C to stop)
MCP_TRANSPORT=http uv run python server.py
test: ## Run the test suite
uv run pytest -v
demo: ## Fetch permanent, expiring, and tampered links against a running server
uv run python scripts/demo.py
deploy: ## Render the launchd plist and start the service
@mkdir -p logs media
@sed -e 's#@UV@#$(UV)#g' \
-e 's#@DIR@#$(DIR)#g' \
-e 's#@PATH@#$(dir $(UV)):/usr/bin:/bin#g' \
deploy/$(LABEL).plist.template > $(PLIST)
launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
launchctl bootstrap $(DOMAIN) $(PLIST)
@echo "Deployed $(LABEL). Set a real MEDIA_SECRET in the plist for production."
undeploy: ## Stop the service and remove the launchd plist
launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
rm -f $(PLIST)
@echo "Removed $(LABEL)."
status: ## Show whether the service is registered and running
@launchctl print $(DOMAIN)/$(LABEL) 2>/dev/null \
| grep -E 'state =|pid =' || echo "$(LABEL) is not loaded."
logs: ## Tail the service log files
@tail -n 40 -f logs/server.out.log logs/server.err.log
clean: ## Remove caches, logs, and stored media
rm -rf .pytest_cache logs media __pycache__ tests/__pycache__
Detailed breakdown
serveruns the HTTP server so links resolve locally during development.deployrenders thelaunchdplist (Step 8) and starts the service, and reminds you to set a realMEDIA_SECRET.demoexercises the links against a running server. Thelaunchdlifecycle (deploy/status/logs/undeploy) matches the companion deploy article.
Step 8: Deploy configuration and a live demo
The plist runs the server over HTTP under launchd with the media environment
set. Add a small demo script that mints links and fetches them.
Create the files
mkdir -p deploy scripts
touch deploy/com.macmcp.media.plist.template scripts/demo.py
Add the code: deploy/com.macmcp.media.plist.template
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
"http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>com.macmcp.media</string>
<key>ProgramArguments</key>
<array>
<string>@UV@</string>
<string>run</string>
<string>python</string>
<string>server.py</string>
</array>
<key>WorkingDirectory</key>
<string>@DIR@</string>
<key>EnvironmentVariables</key>
<dict>
<key>MCP_TRANSPORT</key>
<string>http</string>
<key>MCP_HOST</key>
<string>127.0.0.1</string>
<key>MCP_PORT</key>
<string>8000</string>
<key>MEDIA_DIR</key>
<string>@DIR@/media</string>
<key>MEDIA_BASE_URL</key>
<string>http://127.0.0.1:8000</string>
<key>MEDIA_SECRET</key>
<string>CHANGE-ME-to-a-long-random-string</string>
<key>PATH</key>
<string>@PATH@</string>
</dict>
<key>RunAtLoad</key>
<true/>
<key>KeepAlive</key>
<true/>
<key>StandardOutPath</key>
<string>@DIR@/logs/server.out.log</string>
<key>StandardErrorPath</key>
<string>@DIR@/logs/server.err.log</string>
</dict>
</plist>
Add the code: scripts/demo.py
"""Live demo: generate media, get permanent and expiring links, and fetch them.
Run against a server started with MCP_TRANSPORT=http (see `make serve`).
"""
import asyncio
import os
import urllib.request
from fastmcp import Client
URL = os.environ.get("MCP_URL", "http://127.0.0.1:8000/mcp/")
def _fetch(url: str) -> str:
try:
with urllib.request.urlopen(url) as resp:
return f"HTTP {resp.status} {resp.headers.get('content-type')} {len(resp.read())} bytes"
except urllib.error.HTTPError as exc:
return f"HTTP {exc.code} ({exc.reason})"
async def main() -> None:
async with Client(URL) as client:
perm = (
await client.call_tool(
"badge_svg_link", {"label": "build", "message": "passing"}
)
).data
temp = (
await client.call_tool(
"store_media",
{
"filename": "clip.txt",
"content_base64": "aGVsbG8gdmlkZW8=", # "hello video"
"ttl_minutes": 5,
},
)
).data
print("permanent link:", perm["url"])
print(" fetch:", _fetch(perm["url"]))
print("expiring link :", temp["url"], f"({temp['expires']})")
print(" fetch:", _fetch(temp["url"]))
tampered = temp["url"].rsplit("sig=", 1)[0] + "sig=deadbeef"
print("tampered link :", "sig=deadbeef")
print(" fetch:", _fetch(tampered))
if __name__ == "__main__":
asyncio.run(main())
Detailed breakdown
- The plist sets
MEDIA_DIR,MEDIA_BASE_URL, andMEDIA_SECRET. The base URL must match how clients reach the server so the links it mints are fetchable.MEDIA_SECRETmust be a long random string in production — it is the key that makes expiring links unforgeable; the placeholder is a reminder, not a value to ship. demo.pycalls the link tools, then fetches each URL with the standard library: the permanent and valid-signed links return200with the rightContent-Type, and the tampered link returns403.
Deploy and watch it work:
make deploy
make status # state = running, a live pid
make demo
Expected output:
permanent link: http://127.0.0.1:8000/media/....-badge.svg
fetch: HTTP 200 image/svg+xml 304 bytes
expiring link : http://127.0.0.1:8000/media/....-clip.txt?exp=...&sig=... (in 5 minute(s))
fetch: HTTP 200 text/plain; charset=utf-8 11 bytes
tampered link : sig=deadbeef
fetch: HTTP 403 (Forbidden)
Tear down when done:
make undeploy
Step 9: Move to real object storage (optional)
The local bucket teaches the pattern; production usually wants S3, Cloudflare R2,
or GCS. The mapping is one-to-one, and the tools and the server do not change
— only bucket.py:
store→ upload the bytes (s3.put_object(Bucket, Key, Body)).permanent_url→ the object’s public URL (or a bucket served via CDN).signed_url→ a presigned GET URL with native expiry:s3.generate_presigned_url("get_object", Params={"Bucket": b, "Key": key}, ExpiresIn=ttl_minutes * 60). Presigned URLs are verified by the storage provider, so you can then drop the/mediaroute andverifyentirely — the cloud enforces expiry for you.
Because the tool signatures (-> dict with url/expires) are unchanged, MCP
clients see no difference.
Troubleshooting
- A link returns
403 link expiredimmediately. The server’s clock andttl_minutesinteract; make sure you are not passingttl_minutes=0expecting an expiring link —0means permanent. Any positive value expires. - A link returns
403 bad signatureafter redeploying. TheMEDIA_SECRETchanged, invalidating previously minted expiring links. Keep the secret stable in production (and secret — anyone with it can mint links). - Inline images don’t render in a client. Confirm the client supports image
content blocks.
swatch_pngreturnsimage/png; the SVG helper returnsimage/svg. For guaranteed-correct MIME types, deliver via link — the bucket route setsContent-Typefrom the filename. 404 not foundfor a link you just made. The bucket is a local directory (MEDIA_DIR). If you cleaned it (make clean) or deployed with a differentMEDIA_DIR, the object is gone. Permanent links are only as durable as the directory.- Links point at
127.0.0.1but a client can’t reach them. SetMEDIA_BASE_URLto a hostname the client can resolve; the tools build links from it.
Recap
You built a FastMCP server that delivers media two ways:
- Inline —
Imageand text content returned straight from the tool, for small renderable media. - Link — bytes stored in a bucket, returned as a permanent or
HMAC-signed expiring URL, served by a
/mediaroute on the same server, for large media (video) or shareable results. - Tested end to end — bucket logic, tool contracts, and the HTTP route
(200/403/404) — and deployed as a
launchdservice withuvandmake. - Portable —
bucket.pymaps one-to-one onto S3/R2/GCS presigned URLs without changing the tools.
Next improvements
- Swap
bucket.pyfor S3/R2 presigned URLs (Step 9) and delete the/mediaroute. - Add a
ContentType/size cap and an allowlist of extensions tostore_media. - Add a background sweep that deletes bucket files past a max age (local storage never auto-expires the bytes, only the links).
- Return an MCP resource link for large media so clients can fetch lazily via the protocol instead of a raw URL.