MCP映射资源库
](https://badge.fury.io/py/mcp-mapped-resource-lib) ](https://pypi.org/project/mcp-mapped-resource-lib/)  
一个pip可安装的Python库，为通过共享Docker卷处理二进制blob传输的MCP服务器提供可重用的实用程序。
特性
- Blob 存储：上传、检索、列出和删除具有唯一标识符的二进制文件
- 资源标识符：格式为的唯一blob ID
blob://TIMESTAMP-HASH.EXT - 元数据跟踪：基于JSON的元数据存储以及blob
- 懒惰的清理：基于TTL的自动过期，具有可配置的清理间隔
- 安全：路径遍历阻止、MIME类型验证、大小限制
- 去重：使用SHA256哈希进行可选的基于内容的重复数据消除
- Docker卷支持：专为跨多个MCP服务器共享Docker卷而设计
安装
来自PyPI（推荐）
pip install mcp-mapped-resource-lib系统相关性
图书馆需要 libmagic 对于MIME类型检测：
# Ubuntu/Debian
sudo apt-get install libmagic1
# macOS
brew install libmagic
# Windows (using conda)
conda install -c conda-forge python-magic来源（发展）
git clone https://github.com/nickweedon/mcp_mapped_resource_lib.git
cd mcp_mapped_resource_lib
pip install -e ".[dev]"快速开始
from mcp_mapped_resource_lib import BlobStorage, maybe_cleanup_expired_blobs
# Initialize storage
storage = BlobStorage(
storage_root="/mnt/blob-storage",
max_size_mb=100,
allowed_mime_types=["image/*", "application/pdf"],
enable_deduplication=True
)
# Upload a blob
result = storage.upload_blob(
data=b"Hello, world!",
filename="hello.txt",
tags=["example"],
ttl_hours=24
)
print(f"Blob ID: {result['blob_id']}")
print(f"File path: {result['file_path']}")
print(f"SHA256: {result['sha256']}")
# Retrieve metadata
metadata = storage.get_metadata(result['blob_id'])
print(f"Created: {metadata['created_at']}")
# List blobs with filtering
results = storage.list_blobs(
mime_type="text/*",
tags=["example"],
page=1,
page_size=20
)
print(f"Found {results['total']} blobs")
# Get filesystem path for direct access
file_path = storage.get_file_path(result['blob_id'])
with open(file_path, 'rb') as f:
data = f.read()
# Delete a blob
storage.delete_blob(result['blob_id'])
# Lazy cleanup (run periodically)
cleanup_result = maybe_cleanup_expired_blobs(
storage_root="/mnt/blob-storage",
ttl_hours=24,
cleanup_interval_minutes=5
)
if cleanup_result:
print(f"Deleted {cleanup_result['deleted_count']} expired blobs")
print(f"Freed {cleanup_result['freed_bytes']} bytes")与MCP服务器集成
此库旨在导入MCP服务器（使用FastMCP或其他框架构建）：
from mcp_mapped_resource_lib import BlobStorage, maybe_cleanup_expired_blobs
from fastmcp import FastMCP
import base64
mcp = FastMCP("my-mcp-server")
storage = BlobStorage(storage_root="/mnt/blob-storage")
@mcp.tool()
def upload_blob(
data: str, # base64-encoded
filename: str,
mime_type: str | None = None,
tags: list[str] | None = None
) -> dict:
"""Upload a binary blob and receive a resource identifier."""
binary_data = base64.b64decode(data)
result = storage.upload_blob(
data=binary_data,
filename=filename,
mime_type=mime_type,
tags=tags
)
# Trigger lazy cleanup
maybe_cleanup_expired_blobs(
storage_root="/mnt/blob-storage",
ttl_hours=24
)
return result
@mcp.tool()
def get_blob_content(blob_id: str) -> str:
"""Retrieve blob content as base64-encoded string."""
file_path = storage.get_file_path(blob_id)
with open(file_path, 'rb') as f:
return base64.b64encode(f.read()).decode()
@mcp.tool()
def get_blob_file_path(blob_id: str) -> dict:
"""Get filesystem path for direct blob access (for shared volumes)."""
file_path = storage.get_file_path(blob_id)
return {"blob_id": blob_id, "file_path": str(file_path)}核心模块
blob存储
blob存储操作的主类：
upload_blob()-上传二进制数据并接收资源标识符get_metadata()-检索blob元数据list_blobs()-列出具有过滤和分页功能的blobdelete_blob()-删除blobget_file_path()-获取直接访问的文件系统路径
Blob ID实用程序
用于处理blob标识符的函数：
create_blob_id()-生成唯一的blob标识符validate_blob_id()-验证blob ID格式和安全性parse_blob_id()-将blob ID解析为组件strip_blob_protocol()-删除“blob：//”前缀
清理实用程序
用于管理blob生命周期的函数：
maybe_cleanup_expired_blobs()-带间隔检查的延迟清理cleanup_expired_blobs()-强制清理过期的blobshould_run_cleanup()-检查清理间隔是否已过scan_for_expired_blobs()-查找过期的blob
路径实用程序
路径解析和安全功能：
blob_id_to_path()-将blob ID转换为文件系统路径get_metadata_path()-获取元数据文件路径sanitize_filename()-对用户提供的文件名进行消毒validate_path_safety()-防止路径遍历攻击
MIME和哈希实用程序
内容处理功能：
detect_mime_type()-从数据和文件名中检测MIME类型validate_mime_type()-根据允许的列表验证MIME类型calculate_sha256()-计算SHA256哈希值
配置选项
块存储配置
storage = BlobStorage(
storage_root="/mnt/blob-storage", # Storage directory
max_size_mb=100, # Max blob size in MB
allowed_mime_types=["image/*"], # Allowed MIME types (None = all)
enable_deduplication=True, # Enable SHA256 deduplication
default_ttl_hours=24 # Default TTL for blobs
)清理配置
result = maybe_cleanup_expired_blobs(
storage_root="/mnt/blob-storage",
ttl_hours=24, # Time-to-live in hours
cleanup_interval_minutes=5 # Min interval between cleanups
)目录结构
该库使用两级目录分片来提高性能：
/mnt/blob-storage/
├── 17/ # First 2 digits of timestamp
│ ├── 33/ # Digits 3-4 of timestamp
│ │ ├── 1733437200-a3f9d8c2b1e4f6a7.png
│ │ └── 1733437200-a3f9d8c2b1e4f6a7.png.meta.json
│ └── 34/
├── 18/
└── .last_cleanup # Cleanup tracking file安全功能
- 路径穿越预防：严格的验证正则表达式阻止目录遍历
- MIME类型筛选：允许的MIME类型的可配置白名单
- 大小限制：可配置的最大blob大小
- 输入消毒：文件名在存储前经过净化
- 路径安全验证：确保解析的路径保持在存储根目录内
Docker卷配置
共享卷的Docker Compose配置示例：
services:
mcp-server-1:
build: .
volumes:
- blob-storage:/mnt/blob-storage # Read-write access
environment:
- BLOB_STORAGE_ROOT=/mnt/blob-storage
mcp-server-2:
build: .
volumes:
- blob-storage:/mnt/blob-storage:ro # Read-only access
environment:
- BLOB_STORAGE_ROOT=/mnt/blob-storage
volumes:
blob-storage:
driver: local文档
发展
使用DevContainer（推荐）
此项目包括VS Code的完整DevContainer设置：
- 安装 开发容器扩展
- 在VS Code中打开项目
- 出现提示时，单击“在容器中重新打开”（或从命令面板运行“开发容器：在容器中再次打开”）
DevContainer包括：
- 带uv包管理器的Python 3.12
- 所有开发依赖项均已预安装
- Docker CLI用于测试容器化MCP服务器
- 用于人工智能辅助开发的Claude Code CLI
- 预配置的扩展（Python、Pylance、Ruff、Claude Code）
本地开发
# Install development dependencies
make install
# or: pip install -e ".[dev]"
# Run all checks (recommended - runs lint, typecheck, and test)
make all
# Run individual checks
make lint # Run ruff linting
make typecheck # Run mypy type checking
make test # Run pytest with coverage
# Clean build artifacts
make clean
# Show available commands
make help直接命令（无Makefile）
# Lint code
ruff check src/ tests/
# Type check
mypy src/
# Run tests with coverage
pytest tests/ --cov=mcp_mapped_resource_lib --cov-report=html发布新版本
该项目包括一个简单的发布自动化工作流，用于创建新版本并发布到PyPI。
先决条件
- 安装GitHub CLI (
gh):
# macOS
brew install gh
# Ubuntu/Debian
sudo apt install gh
# Windows
winget install GitHub.cli- 通过GitHub进行身份验证:
gh auth login创建发布
使用Makefile目标创建新版本：
# Auto-increment patch version (e.g., 0.1.0 → 0.1.1)
make release
# Or specify a version explicitly for minor/major releases
make release VERSION=0.2.0或者直接使用脚本：
# Auto-increment patch version
./release.sh
# Or specify version explicitly
./release.sh 0.2.0释放期间会发生什么
发布脚本会自动执行以下步骤：
- 检测 git标签的最新版本（不是
pyproject.toml) - 确认 版本格式（必须是字母：X.Y.Z）
- 检查 GitHub CLI已安装并经过身份验证
- 验证 工作目录是干净的（没有未提交的更改）
- 更新 版本在
pyproject.toml - 跑动
make all（皮棉+类型检查+测试）以确保质量 - 承诺 版本碰撞
- 创建 git标签（例如。，
v0.2.0) - 推动 提交并标记到GitHub
- 创建 GitHub发布，带有自动生成的发布说明
- 触发器 通过GitHub Actions自动发布PyPI
重要说明
- ⚠️ 这
release目标是 不 常规构建过程的一部分(make all) - ⚠️ 必须创建发布 明确地 -它们从来不会自动发生
- ✅ 如果未指定版本，则补丁版本将从 最新git标签 （不是
pyproject.toml) - ✅ 跑
git fetch --tags在发布之前，确保您拥有GitHub上的最新标签 - ✅ 对于次要/主要版本，明确指定版本（例如。，
VERSION=0.2.0或VERSION=1.0.0) - ✅ 必须通过所有测试才能创建版本
- ✅ PyPI发布通过GitHub Actions自动进行（使用受信任的发布者）
- ✅ 您可以在以下位置监视发布工作流：https://github.com/nickweedon/mcp_mapped_resource_lib/actions
版本控制
该项目如下 语义化版本:
- 补丁 （0.0.X）：错误修复，向后兼容-
make release（自动递增） - 次要的 （0.X.0）：新功能，向后兼容-
make release VERSION=0.2.0 - 重大 （X.0.0）：突破性更改-
make release VERSION=1.0.0
故障排除
错误：未安装GitHub CLI
- 安装
gh使用上述说明
错误：GitHub CLI未通过身份验证
- 跑
gh auth login并按照提示进行操作
错误：工作目录有未提交的更改
- 在发布之前提交或隐藏您的更改：
git status
错误：测试失败
- 在创建版本之前修复失败的测试
- 版本更改
pyproject.toml将自动恢复
许可证
MIT许可证-有关详细信息，请参阅许可证文件
更新日志
看 更改日志.md 查看版本历史和发行说明。
贡献
欢迎投稿！请参阅CONTRIBUTING.md了解指南。
支持
- 问题：https://github.com/nickweedon/mcp_mapped_resource_lib/issues
- 文档：https://github.com/nickweedon/mcp_mapped_resource_lib/docs
- PyPI：https://pypi.org/project/mcp-mapped-resource-lib/