使用说明

MCP SPARQL服务器

一个灵活而强大的支持SPARQL的MCP（模型上下文协议）服务器

🌟 概述

MCP SPARQL Server是一个高性能、可配置的服务器，可连接到任何SPARQL端点，并提供增强的功能，包括结果格式化和缓存。它建立在实现模型上下文协议（MCP）的FastMCP框架之上，为AI助手查询语义数据提供无缝接口。

✨ 特性

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

📋 需求

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客户端集成

🔍 用法

基本用法（标准传输）

通过指定SPARQL端点启动服务器：

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

HTTP服务器模式（用于nginx/web集成）

将服务器作为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支持：

在环境文件中配置端点：
```
sudo nano /etc/mcp-sparql/env
```
启动服务：
```
sudo systemctl start sparql-server
```
启动时启用：
```
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())

使用HTTP传输的FastMCP客户端（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"})

⚙️ 配置

命令行参数

Argument	Description	Default
`--endpoint URL`	SPARQL endpoint URL	Required
`--timeout SECONDS`	Request timeout in seconds	30
`--format FORMAT`	Result format (json, simplified, tabular)	json
`--cache-enabled BOOL`	Enable result caching	true
`--cache-ttl SECONDS`	Cache time-to-live in seconds	300
`--cache-max-size SIZE`	Maximum cache size	100
`--cache-strategy STRATEGY`	Cache replacement strategy (lru, lfu, fifo)	lru
`--pretty-print`	Pretty print JSON output	false
`--include-metadata BOOL`	Include query metadata in results	true
`--daemon`	Run as a background daemon	false
`--log-file FILE`	Log file location when running as a daemon	/var/log/mcp-sparql-server.log
`--pid-file FILE`	PID file location when running as a daemon	/var/run/mcp-sparql-server.pid
`--transport TRANSPORT`	Transport type (stdio or http)	stdio
`--host HOST`	Host to bind HTTP server to	localhost
`--port PORT`	Port to bind HTTP server to	8000

环境变量

Variable	Description	Default
`SPARQL_ENDPOINT`	SPARQL endpoint URL	None (required)
`SPARQL_TIMEOUT`	Request timeout in seconds	30
`SPARQL_FORMAT`	Default result format	json
`SPARQL_CACHE_ENABLED`	Enable caching	true
`SPARQL_CACHE_TTL`	Cache time-to-live in seconds	300
`SPARQL_CACHE_MAX_SIZE`	Maximum cache size	100
`SPARQL_CACHE_STRATEGY`	Cache replacement strategy	lru
`SPARQL_PRETTY_PRINT`	Pretty print JSON output	false
`SPARQL_INCLUDE_METADATA`	Include query metadata in results	true
`MCP_TRANSPORT`	Transport type (stdio or http)	stdio
`MCP_HOST`	Host to bind HTTP server to	localhost
`MCP_PORT`	Port to bind HTTP server to	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通用公共许可证v3.0（AGPL-3.0）
商业的：可用于商业或专有用途的专有商业许可

请参阅许可证文件以获取完整的详细信息。

该软件由Temkit Sid Ali为Yet.lu设想和开发，并得到了Claude（Anthropic）、GitHub Copilot/Codex（OpenAI）和GPT-o3（OpenAI。

🚀 路线图

我们对未来有令人兴奋的计划！查看我们的详细路线图看看接下来会发生什么，包括：

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

🤝 贡献

欢迎投稿！请随时提交拉取请求。

分叉存储库
创建功能分支(git checkout -b feature/amazing-feature)
提交您的更改(git commit -m 'Add some amazing feature')
推到分支(git push origin feature/amazing-feature)
打开拉取请求

查看我们的贡献指南了解详细的开发设置和指南。

🙏 致谢

这个项目建立在几个优秀的开源项目之上：

FastMCP -Joel Lowin构建MCP服务器和客户端的高级Python框架
SPARQL包装机 -RDFLib团队为Python开发的SPARQL端点接口
派丹蒂克 -Samuel Colvin和Pydantic团队使用Python类型提示进行数据验证
python守护进程 -用于创建Unix守护进程的库

特别感谢：

这 模型上下文协议（MCP） 开发协议规范的社区
Anthropic 感谢他们在人工智能助手和MCP生态系统方面的工作
这 语义互联网 和 资源描述框架 社区的基础工作
W3C SPARQL规范和标准
AI开发合作伙伴:
- 克劳德（人类学） -架构、代码实现和文档的主要开发助理
- GitHub Copilot/Codex（OpenAI） -代码完成和开发加速
- GPT-o3（OpenAI） -高级推理和问题解决辅助

📬 联系

GitHub问题： https://github.com/yet-market/yet-sparql-mcp-server/issues
网站： https://yet.lu
发展：dev@yet.lu
商业许可：legal@yet.lu