Yet Sparql Mcp Server

MCP SPARQL Server

🌟 概述

MCP SPARQL Server 是一款高性能、可配置的服务，可连接到任意 SPARQL 端点，并提供结果格式化和缓存等增强功能。它基于 FastMCP 框架构建，实现了 Model Context Protocol（MCP），为 AI 助手提供无缝接口，以查询语义数据。

✨ 功能

• 通用端点支持：连接到任意兼容 SPARQL 的端点
• 完整的 SPARQL 支持：执行任何有效的 SPARQL 查询（SELECT、ASK、CONSTRUCT、DESCRIBE）
• 智能结果格式化：
  • 标准 JSON（兼容标准 SPARQL 客户端）
  • 简化 JSON（更便于在应用中使用）
  • 表格格式（可直接用于 UI 表格展示）
• 高性能缓存：
  • 多种缓存策略（LRU、LFU、FIFO）
  • 可配置的 TTL（生存时间）
  • 缓存管理工具
• 灵活的部署选项：
  • 使用 stdio 或 HTTP 传输，以前台模式运行
  • 作为后台守护进程运行
  • 部署为 systemd 服务
  • HTTP 服务模式，可集成 nginx 反向代理
• 全面的配置方式：
  • 命令行参数
  • 环境变量
  • 无硬编码值

📋 要求

• Python 3.8 或更高版本
• SPARQLWrapper 库
• fastmcp 框架
• 用于配置的 pydantic
• 用于后台运行的 python-daemon

🚀 安装

从源代码安装

# Clone the repository
git clone https://github.com/yet-market/yet-sparql-mcp-server.git
cd yet-sparql-mcp-server

# Install dependencies
pip install -r requirements.txt

# Install the package
pip install -e .

从 PyPI 安装

pip install mcp-server-sparql

使用安装脚本

如需完整安装并配置 systemd 服务：

# Download the repository
git clone https://github.com/yet-market/yet-sparql-mcp-server.git
cd yet-sparql-mcp-server

# Run the installation script (as root for systemd service)
sudo ./install.sh

安装程序会自动：

• 设置虚拟环境并安装依赖
• 创建并启动 systemd 服务
• 提供便于管理的脚本
• 配置可与 nginx 集成的 HTTP 服务

快速管理

安装后，使用提供的脚本：

# Check status
./status.sh

# Start/stop services
./start.sh [stdio|http]    # Start service (http is default)
./stop.sh [stdio|http]     # Stop service
./logs.sh [stdio|http]     # View logs

# Examples
./start.sh http            # Start HTTP service for nginx
./logs.sh http             # View HTTP service logs
./test.sh                  # Test server functionality
./stop.sh                  # Stop all services

测试安装

使用内置测试套件测试服务：

# Run comprehensive tests
./test.sh

# Run specific tests
./test.sh endpoint         # Test HTTP connectivity
./test.sh service          # Test service status
./test.sh config           # Test configuration
./test.sh mcp              # Test MCP protocol
./test.sh client           # Test with FastMCP client
./test.sh info             # Show connection info

测试脚本会验证：

• ✅ 服务状态和运行状况
• ✅ HTTP 端点连通性
• ✅ MCP 协议兼容性
• ✅ 配置有效性
• ✅ SPARQL 端点可达性
• ✅ FastMCP 客户端集成

🔍 用法

基本用法（stdio 传输）

指定 SPARQL 端点以启动服务：

python server.py --endpoint https://dbpedia.org/sparql

HTTP 服务模式（用于 nginx/网页集成）

将服务作为 HTTP 服务运行：

# Basic HTTP server
python server.py --transport http --endpoint https://dbpedia.org/sparql

# Custom host and port
python server.py --transport http --host 0.0.0.0 --port 8080 --endpoint https://dbpedia.org/sparql

# Using environment variables
export MCP_TRANSPORT=http
export MCP_HOST=0.0.0.0
export MCP_PORT=8000
export SPARQL_ENDPOINT=https://dbpedia.org/sparql
python server.py

作为守护进程运行

将服务作为后台进程运行（stdio 传输）：

python server.py --endpoint https://dbpedia.org/sparql --daemon \
  --log-file /var/log/mcp-sparql.log \
  --pid-file /var/run/mcp-sparql.pid

将 HTTP 服务作为守护进程运行：

python server.py --transport http --host 0.0.0.0 --port 8000 \
  --endpoint https://dbpedia.org/sparql --daemon \
  --log-file /var/log/mcp-sparql.log \
  --pid-file /var/run/mcp-sparql.pid

使用 Systemd

如果安装时启用了 systemd 支持：

1. 在环境文件中配置端点：

   sudo nano /etc/mcp-sparql/env
   

2. 启动服务：

   sudo systemctl start sparql-server
   

3. 设置为开机启动：

   sudo systemctl enable sparql-server
   

客户端查询示例

启动服务后，可通过任何兼容 MCP 的客户端或 FastMCP 客户端使用该服务：

使用 FastMCP 客户端和 stdio 传输（Python）

import asyncio
from fastmcp.client import Client, PythonStdioTransport

async def query_server():
    # Connect to the server
    transport = PythonStdioTransport(
        script_path="server.py",
        args=["--endpoint", "https://dbpedia.org/sparql"]
    )
    
    async with Client(transport) as client:
        # Execute a SPARQL query
        result = await client.call_tool("query", {
            "query_string": "SELECT * WHERE { ?s ?p ?o } LIMIT 5",
            "format": "simplified"
        })
        
        print(result[0].text)

asyncio.run(query_server())

使用 FastMCP 客户端和 HTTP 传输（Python）

import asyncio
from fastmcp.client import Client, HttpTransport

async def query_server():
    # Connect to HTTP server
    transport = HttpTransport("http://localhost:8000")
    
    async with Client(transport) as client:
        # Execute a SPARQL query
        result = await client.call_tool("query", {
            "query_string": "SELECT * WHERE { ?s ?p ?o } LIMIT 5",
            "format": "simplified"
        })
        
        print(result[0].text)

asyncio.run(query_server())

使用不同格式进行查询

# JSON format (default)
result = await client.call_tool("query", {
    "query_string": "SELECT * WHERE { ?s ?p ?o } LIMIT 5",
    "format": "json"
})

# Tabular format
result = await client.call_tool("query", {
    "query_string": "SELECT * WHERE { ?s ?p ?o } LIMIT 5",
    "format": "tabular"
})

缓存管理

# Get cache statistics
cache_stats = await client.call_tool("cache", {"action": "stats"})

# Clear the cache
cache_clear = await client.call_tool("cache", {"action": "clear"})

⚙️ 配置

命令行参数

参数	描述	默认值
--endpoint URL	SPARQL 端点 URL	必需
--timeout SECONDS	请求超时时间（秒）	30
--format FORMAT	结果格式（json、simplified、tabular）	json
--cache-enabled BOOL	启用结果缓存	true
--cache-ttl SECONDS	缓存存活时间（秒）	300
--cache-max-size SIZE	最大缓存大小	100
--cache-strategy STRATEGY	缓存替换策略（lru、lfu、fifo）	lru
--pretty-print	美化 JSON 输出	false
--include-metadata BOOL	在结果中包含查询元数据	true
--daemon	作为后台守护进程运行	false
--log-file FILE	作为守护进程运行时的日志文件位置	/var/log/mcp-sparql-server.log
--pid-file FILE	作为守护进程运行时的 PID 文件位置	/var/run/mcp-sparql-server.pid
--transport TRANSPORT	传输类型（stdio 或 http）	stdio
--host HOST	HTTP 服务要绑定的主机	localhost
--port PORT	HTTP 服务要绑定的端口	8000

环境变量

变量	描述	默认值
SPARQL_ENDPOINT	SPARQL 端点 URL	无（必需）
SPARQL_TIMEOUT	请求超时时间（秒）	30
SPARQL_FORMAT	默认结果格式	json
SPARQL_CACHE_ENABLED	启用缓存	true
SPARQL_CACHE_TTL	缓存存活时间（秒）	300
SPARQL_CACHE_MAX_SIZE	最大缓存大小	100
SPARQL_CACHE_STRATEGY	缓存替换策略	lru
SPARQL_PRETTY_PRINT	美化 JSON 输出	false
SPARQL_INCLUDE_METADATA	在结果中包含查询元数据	true
MCP_TRANSPORT	传输类型（stdio 或 http）	stdio
MCP_HOST	HTTP 服务要绑定的主机	localhost
MCP_PORT	HTTP 服务要绑定的端口	8000

🌐 Nginx 集成

使用 HTTP 传输时，可以将服务与 nginx 集成，作为反向代理：

1. 以 HTTP 模式启动服务

python server.py --transport http --host localhost --port 8000 --endpoint https://your-sparql-endpoint.com/sparql

2. 配置 nginx

将此配置添加到 nginx 服务块中：

upstream mcp_sparql {
    server localhost:8000;
    # Add more servers for load balancing if needed
    # server localhost:8001;
    # server localhost:8002;
}

server {
    listen 80;
    server_name your-domain.com;

    location /api/sparql {
        proxy_pass http://mcp_sparql;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # Optional: Add CORS headers
        add_header Access-Control-Allow-Origin *;
        add_header Access-Control-Allow-Methods "GET, POST, OPTIONS";
        add_header Access-Control-Allow-Headers "Content-Type, Authorization";
    }
}

3. 使用 systemd 部署到生产环境

为 HTTP 模式创建 systemd 服务：

# /etc/systemd/system/mcp-sparql-http.service
[Unit]
Description=MCP SPARQL Server (HTTP)
After=network.target

[Service]
Type=simple
User=mcp-sparql
Group=mcp-sparql
WorkingDirectory=/opt/mcp-sparql
Environment=MCP_TRANSPORT=http
Environment=MCP_HOST=localhost
Environment=MCP_PORT=8000
Environment=SPARQL_ENDPOINT=https://your-sparql-endpoint.com/sparql
ExecStart=/opt/mcp-sparql/venv/bin/python server.py
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

启用并启动服务：

sudo systemctl enable mcp-sparql-http
sudo systemctl start mcp-sparql-http

📊 结果格式

服务支持三种不同的输出格式：

1. JSON 格式（默认）

返回标准的 SPARQL JSON 结果格式，并可选择包含元数据。

{
  "head": {
    "vars": ["s", "p", "o"]
  },
  "results": {
    "bindings": [
      {
        "s": { "type": "uri", "value": "http://example.org/resource" },
        "p": { "type": "uri", "value": "http://example.org/property" },
        "o": { "type": "literal", "value": "Example Value" }
      }
    ]
  },
  "metadata": {
    "variables": ["s", "p", "o"],
    "count": 1,
    "query": "SELECT * WHERE { ?s ?p ?o } LIMIT 1"
  }
}

2. 简化格式

返回更易于使用的简化 JSON 结构，将变量绑定转换为简单的键值对象。

{
  "type": "SELECT",
  "results": [
    {
      "s": "http://example.org/resource",
      "p": "http://example.org/property",
      "o": "Example Value"
    }
  ],
  "metadata": {
    "variables": ["s", "p", "o"],
    "count": 1,
    "query": "SELECT * WHERE { ?s ?p ?o } LIMIT 1"
  }
}

3. 表格格式

以包含列和行的表格形式返回结果，适合用于表格展示。

{
  "type": "SELECT",
  "columns": [
    { "name": "s", "label": "s" },
    { "name": "p", "label": "p" },
    { "name": "o", "label": "o" }
  ],
  "rows": [
    [
      "http://example.org/resource",
      "http://example.org/property",
      "Example Value"
    ]
  ],
  "metadata": {
    "variables": ["s", "p", "o"],
    "count": 1,
    "query": "SELECT * WHERE { ?s ?p ?o } LIMIT 1"
  }
}

🔄 缓存策略

服务支持三种缓存替换策略：

1. LRU（最近最少使用）

优先淘汰最近最少访问的项目。这是默认策略，适用于大多数场景，因为它会优先将最近访问的项目保留在缓存中。

2. LFU（最不经常使用）

优先淘汰访问频率最低的项目。此策略适用于某些查询远比其他查询常见的场景，因为它会优先将经常访问的项目保留在缓存中。

3. FIFO（先进先出）

无论访问模式如何，都优先淘汰最早加入的项目。此策略较为简单，在需要纯粹基于时间的缓存方式时很有用。

🔍 高级 SPARQL 示例

服务支持所有 SPARQL 功能。以下是一些可尝试的示例查询：

基本三元组模式

SELECT * WHERE { ?s ?p ?o } LIMIT 10

按属性类型筛选

PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>

SELECT ?subject ?label
WHERE {
    ?subject rdf:type rdfs:Class ;
             rdfs:label ?label .
}
LIMIT 10

使用正则表达式

PREFIX foaf: <http://xmlns.com/foaf/0.1/>

SELECT ?person ?name
WHERE {
    ?person foaf:name ?name .
    FILTER(REGEX(?name, "Smith", "i"))
}
LIMIT 10

包含多个模式的复杂查询

PREFIX dbo: <http://dbpedia.org/ontology/>
PREFIX dbp: <http://dbpedia.org/property/>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>

SELECT ?city ?name ?population ?country ?countryName
WHERE {
    ?city a dbo:City ;
          rdfs:label ?name ;
          dbo:population ?population ;
          dbo:country ?country .
    ?country rdfs:label ?countryName .
    FILTER(?population > 1000000)
    FILTER(LANG(?name) = 'en')
    FILTER(LANG(?countryName) = 'en')
}
ORDER BY DESC(?population)
LIMIT 10

⚠️ 故障排除

常见问题

• 连接被拒绝：检查 SPARQL 端点 URL 是否正确且可访问
• 查询超时：使用 --timeout 选项增大超时时间
• 大型结果集导致内存问题：在查询中添加 LIMIT 子句，或减小缓存大小
• 日志/PID 文件权限被拒绝：检查目录权限，或使用适当的权限运行

日志记录

以前台模式运行时，日志会输出到控制台。以守护进程模式运行时，日志会写入指定的日志文件（默认：/var/log/mcp-sparql-server.log）。

如需提高详细程度，可以在源代码中设置 Python 日志级别。

🛠️ 开发

项目结构

mcp-server-sparql/
├── sparql_server/             # Main package
│   ├── core/                  # Core functionality
│   │   ├── __init__.py        # Package exports
│   │   ├── config.py          # Configuration management
│   │   └── server.py          # Main SPARQL server
│   ├── formatters/            # Result formatters
│   │   ├── __init__.py        # Package exports
│   │   ├── formatter.py       # Base formatter class
│   │   ├── json_formatter.py  # JSON formatter
│   │   ├── simplified_formatter.py # Simplified JSON formatter
│   │   └── tabular_formatter.py # Tabular formatter
│   ├── cache/                 # Caching implementation
│   │   ├── __init__.py        # Package exports
│   │   ├── query_cache.py     # Base cache interface
│   │   ├── lru_cache.py       # LRU cache implementation
│   │   ├── lfu_cache.py       # LFU cache implementation
│   │   └── fifo_cache.py      # FIFO cache implementation
│   └── __init__.py            # Package exports
├── server.py                  # Main entry point
├── setup.py                   # Package setup
├── install.sh                 # Installation script
├── requirements.txt           # Python dependencies
├── sparql-server.service      # Systemd service file
├── README.md                  # This file
└── LICENSE                    # License file

运行测试

# Test stdio transport
python test_sparql_server.py

# Test HTTP transport (start server first)
# Terminal 1:
python server.py --transport http --endpoint https://dbpedia.org/sparql
# Terminal 2:
# Run HTTP-specific tests (if available)

🔒 安全注意事项

• 服务不实现身份验证或授权，而是依赖底层 SPARQL 端点的安全性
• 用于生产环境时，请考虑将其部署在安全代理之后
• 请谨慎处理不受信任的查询，因为它们可能会消耗大量资源

📄 许可证

本项目采用双重许可模式：

• 开源：开源使用适用 GNU Affero General Public License v3.0 (AGPL-3.0)
• 商业：商业或专有用途可使用专有商业许可证

详情请参阅 LICENSE 文件。

本软件由 Temkit Sid-Ali 为 Yet.lu 构思并开发，并由 Claude (Anthropic)、GitHub Copilot/Codex (OpenAI) 和 GPT-o3 (OpenAI) 提供 AI 辅助。

🚀 路线图

我们对未来有令人期待的规划！查看详细路线图，了解接下来的计划，包括：

• 🔒 增强安全性和身份验证
• 🌐 用于探索查询的 Web 界面
• 🤖 由 AI 驱动的自然语言转 SPARQL
• 📊 高级数据可视化和分析
• 🏢 企业级功能和可扩展性改进

🤝 参与贡献

欢迎贡献！请随时提交 Pull Request。

1. Fork 此代码库
2. 创建功能分支（git checkout -b feature/amazing-feature）
3. 提交更改（git commit -m 'Add some amazing feature'）
4. 推送到分支（git push origin feature/amazing-feature）
5. 发起 Pull Request

请参阅我们的贡献指南，了解详细的开发环境设置和规范。

🙏 致谢

本项目基于多个优秀的开源项目构建：

• FastMCP - Joel Lowin 开发的高级 Python 框架，用于构建 MCP 服务和客户端
• SPARQLWrapper - RDFLib 团队开发的 Python SPARQL 端点接口
• Pydantic - Samuel Colvin 和 Pydantic 团队开发的使用 Python 类型提示进行数据验证的工具
• python-daemon - 用于创建 Unix 守护进程的库

特别感谢：

• Model Context Protocol (MCP) 社区开发了该协议规范
• Anthropic 在 AI 助手和 MCP 生态系统方面所做的工作
• 语义网和RDF社区所做的奠基性工作
• W3C 制定了 SPARQL 规范和标准
• AI 开发伙伴：
  • Claude (Anthropic) - 主要开发助手，负责架构设计、代码实现和文档编写
  • GitHub Copilot/Codex (OpenAI) - 代码补全和加速开发
  • GPT-o3 (OpenAI) - 提供高级推理和问题解决协助

📬 联系方式

• GitHub Issues：https://github.com/yet-market/yet-sparql-mcp-server/issues
• 网站：https://yet.lu
• 开发联系：[email protected]
• 商业许可：[email protected]

---