bps-searcher 0.1.4

BPS搜索MCP工具 - 用于查询印尼相关数据

BPS搜索MCP工具

项目概述

本项目利用BPS的WEB-API能力，将其封装为一个MCP工具，方便其他大模型调用来查询印尼相关数据。

功能特性

• 支持19种不同的数据模型查询
• 提供完整的错误处理和异常情况处理
• 基于FastMCP框架实现的MCP服务器
• 支持日志记录和调试
• 内置数据缓存机制，提高查询性能
• 增强的分页功能，方便处理大量数据
• 完整的单元测试和集成测试覆盖
• 支持异步测试和网络错误模拟测试

安装和使用

安装依赖

uv sync

运行MCP服务器

uv run -m src.bps_searcher

在Claude Code中配置

claude mcp add bps --scope local --env BPS_APP_ID=xxxx \
  --  uv --directory /$PWD/bps-searcher run -m src.bps_searcher

运行测试

uv run pytest

运行特定测试文件：

uv run pytest tests/test_bps_api_client.py

运行测试并显示详细结果：

uv run pytest -v

支持的数据模型

• subject: 主题
• subcat: 子主题
• data: 动态数据
• truth: 衍生周期
• turvar: 派生变量
• th: 周期
• unit: 单位
• var: 变量
• vervar: 垂直变量
• statictable: 静态表格
• subcatcsa: 统计活动分类
• pressrelease: 新闻稿
• publication: 出版物
• indicators: 指标
• infographic: 信息图标
• glosarium: 词汇表
• sdgs: 可持续发展
• sdds: 特殊数据发布标准
• news: 新闻

使用示例

搜索新闻数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

# 创建API客户端，设置缓存时间为600秒（10分钟）
client = BPSAPIInterface("your_app_id", cache_ttl=600)
result = client.search_news(domain="0000", page=1)
print(result)

使用分页功能搜索数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")

# 使用分页功能搜索新闻数据
pagination_result = client.search_with_pagination(client.search_news, domain="0000", page=1)

print(f"当前页: {pagination_result.pagination.page}")
print(f"总页数: {pagination_result.pagination.pages}")
print(f"是否有下一页: {pagination_result.has_next}")
print(f"数据记录数: {len(pagination_result.data)}")

# 如果有下一页，获取下一页数据
if pagination_result.has_next:
    next_page = pagination_result.get_next_page()
    pagination_result2 = client.search_with_pagination(client.search_news, domain="0000", page=next_page)
    print(f"第二页数据记录数: {len(pagination_result2.data)}")

使用关键词搜索功能

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")

# 使用关键词搜索新闻数据
result = client.search_news(domain="0000", page=1, keyword="镍矿")
print(f"关键词搜索结果: {len(result.data)}条记录")

指定语言进行搜索

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")

# 使用英语搜索主题数据
result = client.search_subjects(domain="0000", page=1, lang="eng")
print(f"英语搜索结果: {len(result.data)}条记录")

搜索主题数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_subjects(domain="0000", page=1)
print(result)

搜索子主题数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_subcat(domain="0000", page=1)
print(result)

搜索动态数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_data(domain="0000", page=1)
print(result)

搜索衍生周期数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_truth(domain="0000", page=1)
print(result)

搜索派生变量数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_turvar(domain="0000", page=1)
print(result)

搜索周期数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_periods(domain="0000", page=1)
print(result)

搜索单位数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_units(domain="0000", page=1)
print(result)

搜索变量数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_variables(domain="0000", page=1)
print(result)

搜索区域变量数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_regions(domain="0000", page=1)
print(result)

搜索静态表格数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_statictable(domain="0000", page=1)
print(result)

搜索统计活动分类数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_subcatcsa(domain="0000", page=1)
print(result)

搜索新闻稿数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_pressrelease(domain="0000", page=1)
print(result)

搜索出版物数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_publications(domain="0000", page=1)
print(result)

搜索指标数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_indicators(domain="0000", page=1)
print(result)

搜索信息图标数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_infographic(domain="0000", page=1)
print(result)

搜索词汇表数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_glosarium(domain="0000", page=1)
print(result)

搜索可持续发展数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_sdgs(domain="0000", page=1)
print(result)

搜索特殊数据发布标准数据

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
result = client.search_sdds(domain="0000", page=1)
print(result)

使用MCP工具搜索主题数据

# 通过MCP客户端调用search_subjects工具
result = mcp_client.call_tool("search_subjects", {"domain": "0000", "page": 1})

使用MCP工具进行高级搜索

# 通过MCP客户端调用advanced_search工具
result = mcp_client.call_tool("advanced_search", {
    "query": "经济",
    "model": "news",
    "domain": "0000",
    "page": 1
})

指定领域ID进行搜索

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")
# 搜索Aceh省的新闻数据
result = client.search_news(domain="1100", page=1)
print(result)

API调用错误处理示例

API客户端调用错误处理

from src.bps_searcher.bps_api_client import BPSAPIInterface

client = BPSAPIInterface("your_app_id")

try:
    result = client.search_news(domain="0000", page=1)
    if result.status == "OK":
        print(f"成功获取数据，共{len(result.data)}条记录")
    else:
        print(f"API返回错误: {result.status}")
except Exception as e:
    print(f"网络请求失败: {str(e)}")

MCP工具调用错误处理

try:
    # 通过MCP客户端调用search_subjects工具
    result = mcp_client.call_tool("search_subjects", {"domain": "0000", "page": 1})
    if result["status"] == "OK":
        print(f"成功获取数据，共{len(result['data'])}条记录")
    else:
        print(f"工具调用失败: {result['message']}")
except Exception as e:
    print(f"MCP工具调用失败: {str(e)}")

项目结构

bps-searcher/
├── pyproject.toml         # 项目依赖配置
├── README.md              # 项目说明文档
├── PLAN.md                # 项目计划
├── SUMMARY.md             # 项目总结
├── src/                   # 源代码目录
│   └── bps_searcher/      # BPS搜索器包
│       ├── __init__.py    # 包初始化文件
│       ├── bps_api_client.py  # BPS API客户端实现
│       ├── bps_mcp_server.py  # MCP服务器实现
│       └── main.py        # 主程序入口
├── tests/                 # 测试目录
│   ├── test_bps_api.py    # API基础测试脚本
│   ├── test_bps_api_client.py # API客户端增强测试
│   ├── test_bps_api_enhanced.py # API增强功能测试
│   └── test_bps_mcp.py    # MCP测试脚本
├── docs/                  # 文档目录
|   ├── archive/               # 归档文档，已过期文档
│   ├── API_DOCUMENTATION.md   # API数据模型说明
│   └── BPS_WEB_API_DOC.md     # BPS WEB API文档
│   └── CHANGELOG.md           # CHANGELOG文档
├── examples/              # 示例目录
│   ├── data/              # 示例数据文件
│   ├── resources/         # 资源文件
│   └── get_examples.py    # 示例获取脚本
└── uv.lock                # 依赖锁定文件

配置

• APP_ID: BPS API的鉴权KEY，必须通过环境变量BPS_APP_ID进行配置
• 缓存: API客户端内置数据缓存机制，默认缓存时间为300秒（5分钟）

安全性改进

为了提高安全性，我们移除了默认的APP_ID值，强制用户自己设置环境变量。如果未设置BPS_APP_ID环境变量，MCP服务器将拒绝启动并显示明确的错误信息。

请按照以下步骤设置环境变量：

export BPS_APP_ID=your_app_id_here

或者在Windows系统中：

set BPS_APP_ID=your_app_id_here

注意：请将your_app_id_here替换为您从BPS官方网站获取的实际APP_ID。

领域ID说明

领域ID用于指定搜索的数据范围：

• "0000": 全国数据（默认值）
• "1100": Aceh省
• "1200": Sumatera Utara省
• "1300": Sumatera Barat省
• ...
• 完整的领域ID列表请参考 examples/resources/resource/domain-all.json 文件

API文档

详细的API数据模型说明请参考 docs/API_DOCUMENTATION.md 文件。

测试

本项目包含完整的测试套件，涵盖了：

• 单元测试：测试各个API客户端方法
• 集成测试：测试MCP服务器功能
• 网络错误模拟测试：测试网络异常情况处理
• 异步测试：测试异步功能

运行测试：

uv run pytest

更新日志

详细的项目变更历史请参考 docs/CHANGELOG.md 文件。

后续开发计划

1. 进一步完善文档和使用示例