multimodal-mcp-router

An MCP server that exposes multimodal tools (text search, image metadata and captions, audio analysis, video frame sampling) over a local media store, plus a CrewAI multi-agent client in which a Router agent identifies the modality of the user's request, hands it to the matching Specialist agent (which gathers context through the MCP tools), and a Responder agent writes the final answer.

Why this shape: one MCP server can serve many clients (Claude Desktop, IDE assistants, agent frameworks) with the same tools, and routing by modality keeps each specialist's tool list small and its prompts focused. The server is deliberately built on lightweight dependencies (scikit-learn, Pillow, the wave module, optional imageio/opencv) so it runs anywhere; the same interface can be backed by Pixeltable when you need real multimodal data infrastructure (see below).

Everything except the LLM-driven crew runs offline: python run_client.py --demo starts the server in-process over stdio, routes three sample queries with a deterministic router, calls the tools and prints answers - no API key needed.

Architecture

flowchart LR
    U[User query] --> R{"Router agent<br/>text / image / audio / video<br/>fallback: keyword + extension heuristics"}
    R -->|text| TS[Text Specialist]
    R -->|image| IS[Image Specialist]
    R -->|audio| AS[Audio Specialist]
    R -->|video| VS[Video Specialist]
    TS & IS & AS & VS --> RESP["Responder agent<br/>writes the final answer"]
    RESP --> A[Answer]

    subgraph MCP["MCP server (FastMCP, stdio or streamable-http)"]
        T1["search_text<br/>TF-IDF"]
        T2["image_info<br/>image_caption"]
        T3[audio_info]
        T4[video_frames]
        T5[list_media]
    end
    TS <--> T1
    IS <--> T2
    AS <--> T3
    VS <--> T4
    TS & IS & AS & VS <--> T5
    MCP --> STORE[("Media store<br/>FilesystemStore over media/<br/>or PixeltableStore")]

The specialists talk to the server through CrewAI's native MCP support (Agent(mcps=[MCPServerStdio(...)]), CrewAI 1.x): CrewAI launches the stdio server, lists its tools, converts them into CrewAI tools and applies a per-agent tool filter, so the image specialist only ever sees image_info, image_caption and list_media. The pipeline itself is a CrewAI Flow whose @router step emits the modality label and whose @listen handlers run a two-task crew (specialist task, then responder task).

Tools

| Tool | Arguments | What it returns | Implementation | | --- | --- | --- | --- | | search_text | query, top_k=3 | ranked document paths, scores, snippets | scikit-learn TfidfVectorizer over the store's text files | | image_info | path | width/height, mode, format, EXIF tags, dominant colours (hex, name, share) | Pillow (getexif, median-cut quantisation) | | image_caption | path | caption + its source | sidecar <name>.txt if present, else a metadata-based description; a vision model plugs in via caption_model= | | audio_info | path | duration, sample rate, channels, RMS, peak, dominant frequency | stdlib wave + NumPy FFT for .wav; mutagen (optional) for other formats | | video_frames | path, n=4 | frame count, fps, n evenly spaced frames saved as PNGs with dominant colours | imageio (+imageio-ffmpeg) or opencv-python; structured error if neither is installed | | list_media | - | every file with modality and size | the store |

Paths are relative to the media root; the filesystem store refuses paths that escape it. Errors come back as {"error": ..., "message": ...} rather than exceptions, so an agent can recover (for example by calling list_media).

Code map

multimodal-mcp-router/
├── run_server.py            # `python run_server.py --transport stdio|streamable-http [--port] [--media-root] [--store]`
├── run_client.py            # `python run_client.py "<query>"` (CrewAI) | `--demo` (offline) | `--modality`, `--json`
├── server/
│   ├── mcp_server.py        # FastMCP wiring: build_server(store) registers the six tools; CLI with --transport
│   ├── tools.py             # pure tool functions (search_text, image_info, image_caption, audio_info, video_frames, list_media)
│   ├── store.py             # MediaStore interface; FilesystemStore (default); PixeltableStore (import-guarded reference)
│   └── make_samples.py      # generates media/: PNGs (Pillow), a 440 Hz WAV (NumPy), a 12-frame MP4 (imageio), 3 text docs
├── client/
│   ├── router.py            # deterministic modality router: file extensions > known file names > weighted keywords
│   ├── mcp_client.py        # MCPToolClient: sync facade that spawns the server over stdio (worker thread + queue)
│   ├── agents.py            # CrewAI agents: Router, Text/Image/Audio/Video Specialists (mcps=MCPServerStdio + tool filter), Responder
│   ├── crew.py              # MultimodalPipeline Flow: classify -> @router -> specialist crew -> responder; run_query()
│   └── demo.py              # offline pipeline: deterministic router + scripted specialist plans + template responder
├── media/                   # tiny generated fixtures (images/, audio/, video/, docs/) - regenerate with `python -m server.make_samples`
├── tests/
│   ├── test_tools.py        # each tool on the sample media, path escaping, caption fallbacks, sample generation
│   ├── test_router.py       # routing heuristics, extension priority, known-file mentions, tie-breaking
│   └── test_mcp_roundtrip.py# stdio round trip (list tools, call tools, error payloads) and the offline demo
├── requirements.txt
├── .env.example             # OPENAI_API_KEY / MODEL placeholders (CrewAI path only)
├── .gitignore
└── README.md

How the pieces fit: server/tools.py contains the logic and depends only on a MediaStore; server/mcp_server.py wraps it in FastMCP tools; client/ consumes those tools either through CrewAI's MCP integration (agents.py, crew.py) or through the thin MCPToolClient (demo.py, tests). The router heuristics in client/router.py are shared by the demo and by the Router agent as its fallback.

Run it

pip install -r requirements.txt
python -m server.make_samples          # (re)generate media/ fixtures; they are also committed

# 1. Offline demo - server + tools + routing, no LLM
python run_client.py --demo
python run_client.py --demo --query "Grab a few frames from the colour_sweep video"

# 2. The MCP server on its own
python run_server.py                                   # stdio
python run_server.py --transport streamable-http --port 8000   # http://127.0.0.1:8000/mcp

# 3. The CrewAI pipeline (needs a key: cp .env.example .env and fill OPENAI_API_KEY)
export OPENAI_API_KEY=sk-...        # or put it in .env
python run_client.py "What colours dominate the sunset_gradient image and what does it show?"
python run_client.py --modality audio "Tell me about the tone clip" --json

python -m pytest                     # 26 tests, ~6 s

MODEL selects the LLM for every agent (default gpt-4o-mini; any model id CrewAI/LiteLLM understands). MEDIA_ROOT / --media-root point the server at another directory; --store pixeltable switches the backend.

Pointing MCP clients at the server

Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "multimodal-media": {
      "command": "python",
      "args": ["/absolute/path/to/multimodal-mcp-router/run_server.py", "--transport", "stdio"],
      "env": {"MEDIA_ROOT": "/absolute/path/to/multimodal-mcp-router/media"}
    }
  }
}

Any streamable-HTTP client (Claude Code, the mcp Python SDK, Cursor, etc.): start python run_server.py --transport streamable-http --port 8000 and connect to http://127.0.0.1:8000/mcp. With the Python SDK:

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()
        print([t.name for t in (await session.list_tools()).tools])

CrewAI (already wired in client/agents.py):

from crewai import Agent
from crewai.mcp import MCPServerStdio, create_static_tool_filter

Agent(role="Image Specialist", goal="...", backstory="...",
      mcps=[MCPServerStdio(command="python", args=["run_server.py"],
                           tool_filter=create_static_tool_filter(allowed_tool_names=["image_info", "image_caption"]))])

Plugging in Pixeltable

server/store.py defines the MediaStore interface (list_media, resolve, text_documents). FilesystemStore walks a directory; PixeltableStore is a reference implementation over a Pixeltable table with path, kind, file and text columns (pip install pixeltable, then PixeltableStore.create_table() and --store pixeltable). The import is guarded, so the server never needs Pixeltable installed.

Pixeltable is where this design becomes more than a file walker: its tables have native image / video / audio / document column types, computed columns that run models (captioning, embeddings, object detection) incrementally as rows arrive, views that iterate over video frames or document chunks, and embedding indexes for similarity search. In that setup:

• image_caption reads a computed caption column instead of a sidecar file;
• search_text becomes a vector search over an embedding index;
• video_frames queries a FrameIterator view instead of decoding on the fly;
• the MCP tools stay exactly the same for every client.

Where a vision model plugs in

image_caption resolves in order: an injected caption_model(path) -> str callable, a sidecar text file, then a metadata description. To use a real model, pass caption_model= in server/mcp_server.py (for example a BLIP pipeline from transformers, a hosted multimodal API, or a Pixeltable computed column). The stub keeps the tool contract identical so agents and tests do not change.

Sample output

python run_client.py --demo (actual run, ~1.7 s, no LLM):

MCP server up over stdio; tools: search_text, image_info, image_caption, audio_info, video_frames, list_media
media store: 8 files -> audio/tone_440hz.wav, docs/audio_basics.txt, docs/mcp_overview.txt, docs/pixeltable_notes.txt, images/blue_square.png, images/sunset_gradient.png, images/sunset_gradient.txt, video/colour_sweep.mp4

==============================================================================
QUERY      : What colours dominate the sunset_gradient image and what does it show?
ROUTER     : image (confidence 0.64) - mentions known image file 'images/sunset_gradient.png'
SPECIALIST : image-specialist
  -> image_caption(path='images/sunset_gradient.png')
     {"path": "images/sunset_gradient.png", "caption": "A synthetic sunset: an orange-to-purple gradient sky over a dark horizon with a pale yellow sun.", "source": "sidecar:sunset_gradient.txt"}
  -> image_info(path='images/sunset_gradient.png')
     {"path": "images/sunset_gradient.png", "format": "PNG", "mode": "RGB", "width": 160, "height": 100, "exif": {}, "dominant_colours": [{"hex": "#bb6b45", "name": "orange", "share": 0.302}, {"hex": "#191423", "name": "black", "share": 0.273}, ...
RESPONDER  : A synthetic sunset: an orange-to-purple gradient sky over a dark horizon with a pale yellow sun. The file images/sunset_gradient.png is a 160x100 PNG in RGB mode; dominant colours: orange (51%), black (29%), red (20%). (caption source: sidecar:sunset_gradient.txt)

==============================================================================
QUERY      : How long is the tone_440hz audio clip and what is its sample rate?
ROUTER     : audio (confidence 0.76) - mentions known audio file 'audio/tone_440hz.wav'
SPECIALIST : audio-specialist
  -> audio_info(path='audio/tone_440hz.wav')
     {"path": "audio/tone_440hz.wav", "container": "wav", "channels": 1, "sample_width_bytes": 2, "sample_rate_hz": 16000, "frames": 16000, "duration_s": 1.0, "rms": 0.3415, "peak": 0.5, "dominant_frequency_hz": 440.0}
RESPONDER  : audio/tone_440hz.wav lasts 1.0 s at 16000 Hz with 1 channel(s); RMS level 0.3415, peak 0.5, dominant frequency 440.0 Hz.

==============================================================================
QUERY      : Search the docs: what does Pixeltable do with computed columns?
ROUTER     : text (confidence 1.0) - keyword 'docs' -> text (+1.5)
SPECIALIST : text-specialist
  -> search_text(query='Search the docs: what does Pixeltable do with computed columns?', top_k=3)
     {"query": "...", "results": [{"path": "docs/pixeltable_notes.txt", "score": 0.2572, "snippet": "Pixeltable is multimodal data infrastructure for AI applications. Tables have native image, video, audio and document column types. Computed columns run models"}], "documents_indexed": 4}
RESPONDER  : Best match: docs/pixeltable_notes.txt (score 0.2572): "Pixeltable is multimodal data infrastructure for AI applications. Tables have native image, video, audio and document column types. Computed columns run models"

A video query (--query "Grab a few frames from the colour_sweep video") routes to the video specialist, which calls video_frames and reports 12 frames at 6.0 fps (imageio); sampled 3 frames - frame 0: green; frame 6: pink; frame 11: pink.

With an API key, python run_client.py "<query>" runs the same steps with the LLM agents: the Router agent classifies (the heuristic result is offered to it as a hint and used as a fallback), the specialist decides which MCP tools to call, and the Responder writes prose. Without a key the command exits with a clear message pointing to --demo.

What was verified without an LLM

• The stdio and streamable-HTTP transports (tool listing and calls).
• CrewAI's MCP integration: building the specialist agents discovers exactly the filtered tools (image_info, image_caption, list_media for the image specialist, and so on) and those tool objects execute against the server. The LLM-driven Flow requires OPENAI_API_KEY and was not run here.

Limitations

• search_text is TF-IDF over a handful of files; for real corpora use an embedding index (Pixeltable makes that a table operation).
• image_caption without a model is metadata only; audio_info covers WAV natively and other formats only with mutagen; video_frames needs a decoding backend.
• The keyword router is English-only and intentionally simple; the LLM router handles phrasing it does not cover, and --modality forces a route.