Data Filter MCP

数据过滤器mcp

本地MCP服务器，注册受限的Python过滤器并在本地运行它们 json, yaml，以及 txt 文件夹。

它的作用

• register_filter 只接受一个顶级函数的Python源代码： def filter_item(data):
• run_filter 加载本地文件，将加载的文档传递到 filter_item(data)，并返回以下文本 result_text
• convert_file 加载本地文件，并将其传递给 filter_item(data)，并将返回的文本写入另一个本地文件
• 已注册的筛选器仅存在于内存中，并根据服务器TTL设置自动过期

可以使用什么过滤代码

过滤器主体根据白名单进行AST验证。除了一套精心策划的内置组件(len, sorted, max, min, range, enumerate, zip, sum, any, all、转换等）和安全字符串/字典/列表方法，过滤器还可以使用一组精心策划的标准库模块。模块通过其规范名称公开(math, datetime, hashlib等等）。文件系统、进程、网络和不安全的序列化模块(os, pathlib, shutil, subprocess, socket, urllib, pickle等等）故意不可用。

• lambda 表达 --通常为 key= 论点，例如。 sorted(data, key=lambda item: item.get("score")).Lambda主体的验证规则与过滤器的其他部分相同。
• json — json.loads, json.dumps.
• yaml — yaml.safe_load, yaml.safe_dump.不安全 yaml.load / yaml.dump 故意不暴露。
• re — re.match, re.search, re.fullmatch, re.findall, re.sub, re.subn, re.compile, re.escape，加 Match / Pattern 方法(group, groups, groupdict, start, end, span).
• math --数字助手，例如 math.ceil, math.floor, math.sqrt, math.log, math.exp, math.pow, math.factorial, math.gcd, math.lcm, math.isfinite, math.isclose.
• statistics --骨料，如 statistics.mean, statistics.median, statistics.stdev, statistics.variance, statistics.quantiles.
• datetime — datetime.datetime.fromisoformat, datetime.datetime.now, datetime.timedelta, datetime.timezone.utc，以及实例方法，例如 isoformat, strftime, timestamp, weekday, total_seconds.General实例属性读取如下 dt.year 和 dt.month 不受当前政策的支持。
• decimal — decimal.Decimal(...), quantize, normalize, to_eng_string, to_integral_value.
• collections — collections.Counter, collections.defaultdict, collections.OrderedDict, collections.deque，加上以下方法 most_common, elements, popleft, appendleft, rotate.
• itertools — chain, chain.from_iterable, islice, takewhile, dropwhile, groupby, starmap, accumulate, combinations, permutations, product, filterfalse.
• functools — reduce, partial, cmp_to_key, wraps缓存装饰器，例如 lru_cache 和 cache 有意不公开，因为它们可以在过滤器调用中保留进程本地状态。
• operator — itemgetter, methodcaller，以及算术/比较助手，例如 add, mul, lt, eq, gt. attrgetter 故意不暴露。
• textwrap — fill, wrap, shorten, indent, dedent.
• html — html.escape, html.unescape.
• base64 — b64encode, b64decode, urlsafe_b64encode, urlsafe_b64decode, b32encode, b32decode, b16encode, b16decode.
• hashlib — hashlib.sha256, hashlib.sha1, hashlib.md5, hashlib.blake2b, hashlib.new，加上哈希对象方法，如 hexdigest, digest, update.
• ipaddress — ip_address, ip_network, ip_interface, IPv4Network, IPv6Network，加上以下方法 supernet, subnets, hosts, overlaps, subnet_of, supernet_of.General实例属性读取如下 addr.is_private 和 addr.compressed 不受当前政策的支持。
• unicodedata — category, name, lookup, numeric, digit, decimal, bidirectional, combining, mirrored.
• difflib — get_close_matches, ndiff, unified_diff, context_diff, SequenceMatcher.

注： re.compile 根据过滤器代码提供的模式运行，因此病理模式可能会使服务器停滞（ReDoS）。一些助手，如 difflib.SequenceMatcher 在大输入上也可能占用大量CPU。将筛选器源视为受信任但受限制的。

用uvx跑步

发布到PyPI后，使用以下命令启动服务器：

uvx data-filter-mcp --filter-ttl-seconds 3600 --cleanup-interval-seconds 60

显示可用的CLI标志：

uvx data-filter-mcp --help

限制文件访问 --workdir

默认情况下，服务器可以读取本地文件系统上的任何文件。使用一个或 更多 --workdir 用于将文件读取限制到特定目录的标志：

uvx data-filter-mcp \
  --filter-ttl-seconds 3600 \
  --cleanup-interval-seconds 60 \
  --workdir /Users/me/project \
  --workdir /tmp/data

规则：

• 每 --workdir 值必须是 绝对路径 到现有目录。
• run_filter 将只接受位于允许目录内的文件。
• 如果不 --workdir 提供标志，不应用任何限制（向后兼容）。
• convert_file 总是需要至少一个 --workdir 因为它会写入磁盘。
• convert_file 要求目标路径位于允许的工作目录内。
• convert_file 自动创建缺少的目标父目录。
• convert_file 拒绝替换现有目标文件，除非 overwrite 是 true.

使用写入转换后的文件 convert_file

使用 convert_file 当过滤后的输出应该被持久化而不是返回时 与模型内联。该工具接受：

• filter_id --由返回的标识符 register_filter
• source_file_path --要加载的json/yaml/txt文件的绝对路径
• destination_file_path --应保存返回文本的绝对路径
• file_type --可选源文件类型覆盖(json, yaml，或 txt)
• overwrite --可选布尔值，默认值 false

示例流程：

def filter_item(data):
    return "\n".join(data["items"])

然后打电话 convert_file 来源如下 /tmp/data/items.json 以及a 目的地，如 /tmp/data/out/items.txt。结果以UTF-8编写 文本。返回的元数据包括解析的源和目的地路径， 有效源文件类型， bytes_written，以及现有文件是否 覆盖。

MCP客户端配置示例：

{
  "mcpServers": {
    "data-filter": {
      "command": "uvx",
      "args": [
        "data-filter-mcp",
        "--filter-ttl-seconds",
        "3600",
        "--cleanup-interval-seconds",
        "60",
        "--workdir",
        "/Users/me/project",
        "--workdir",
        "/tmp/data"
      ]
    }
  }
}

在本地运行

python server.py --filter-ttl-seconds 3600 --cleanup-interval-seconds 60
python -m data_filter_mcp.server --filter-ttl-seconds 3600 --cleanup-interval-seconds 60
.venv/bin/data-filter-mcp --filter-ttl-seconds 3600 --cleanup-interval-seconds 60