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/sparqlHTTP服务器模式(用于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/sparql2.配置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:
PREFIX rdfs:
SELECT ?subject ?label
WHERE {
?subject rdf:type rdfs:Class ;
rdfs:label ?label .
}
LIMIT 10使用正则表达式
PREFIX foaf:
SELECT ?person ?name
WHERE {
?person foaf:name ?name .
FILTER(REGEX(?name, "Smith", "i"))
}
LIMIT 10具有多种模式的复杂查询
PREFIX dbo:
PREFIX dbp:
PREFIX rdfs:
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://yet.lu
- 发展:dev@yet.lu
- 商业许可:legal@yet.lu
______________________________________________________________________
Built with ❤️ by Temkit Sid-Ali for Yet.lu
Co-developed with Claude (Anthropic), GitHub Copilot/Codex & GPT-o3 (OpenAI)
© 2025 Yet.lu - All rights reserved
