AI导哈

OSM数据查询工具

一个基于Python的工具，通过Overpass API查询OpenStreetMap数据，并通过MCP和轻量级REST端点提供数据访问，支持异步查询和JSON序列化。

  • by SvenST89
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 SvenST89/osint_mcp_experiment

使用说明

天桥MCP工具

该项目提供了一个基于Python的工具，通过Overpass API查询OpenStreetMap数据，并通过MCP（模型上下文协议）和轻量级REST端点公开数据以进行测试或集成。它允许通过基于标签的过滤和边界框或基于区域的查询对OSM节点、方式和可选关系进行异步查询。几何体被解析为GeoPandas对象，并作为完全JSON可序列化的Pydantic模型返回。

程序结构

  • OverpassQuery：构建并执行Overpass API查询。支持通过JSON或CSV输出按标签、边界框或区域名称进行过滤。可以将几何图形解析为Shapely对象，并处理重试和覆盖服务器可用性。

  • AsyncOverpassClient：使用异步执行多个Overpass查询 aiohttp 具有并发控制。

  • OverpassStructuredTool：包装Overpass客户端并提供 query_region 方法返回结构化 OverpassFeature Pydantic对象。既可以用作MCP工具，也可以用于REST端点。

  • MCP服务器：用途 FastMCP 通过MCP暴露工具。这 query_region 方法已注册 @mcp.tool().支持 stateless_http 流式代理呼叫的传输。

  • REST端点：基于Starlette的端点(/query_region)提供了一个轻量级的HTTP接口，用于测试或集成。包括对几何图形和标签的完整JSON净化。

可运行服务器示例

# Instantiate MCP server
mcp = FastMCP(
    name="Overpass MCP Server",
    description="Overpass OSM Query Tool with Structured Output",
    stateless_http=True,
    host="127.0.0.1",
    port=8008
)

# Instantiate tool logic
client = AsyncOverpassClient(max_concurrent=3)
overpass_tool = OverpassStructuredTool(client)

# MCP tool registration
@mcp.tool()
async def query_region(params: OverpassQueryParams, ctx: Context):
    return await overpass_tool.query_region(params, ctx)

# REST endpoint setup
async def rest_query_region(request: Request):
    payload = await request.json()
    params = OverpassQueryParams(**payload)
    result = await overpass_tool.query_region(params, ctx=None)
    safe_result = sanitize_obj(result.model_dump())  # fully JSON-serializable
    return JSONResponse(safe_result, status_code=200)

rest_app = Starlette(routes=[Route("/query_region", rest_query_region, methods=["POST"])])

# Run both MCP and REST servers concurrently
async def main():
    from uvicorn import Config, Server
    mcp_task = asyncio.create_task(mcp.run_streamable_http_async())
    rest_server = Server(Config(app=rest_app, host="127.0.0.1", port=8010, log_level="info"))
    rest_task = asyncio.create_task(rest_server.serve())
    await asyncio.gather(mcp_task, rest_task)

if __name__ == "__main__":
    print("🚀 Overpass MCP running on 127.0.0.1:8008")
    print("🌍 REST API available on 127.0.0.1:8010/query_region")
    asyncio.run(main())

经验教训（到目前为止…）

JSON序列化和形状几何

Shapely对象不能直接以JSON格式返回。所有几何图形都必须使用以下格式转换为GeoJSON字典 mapping(geom).

坐标验证

OSM数据可能包含无效坐标(NaN, Infinity)或非常大的浮子。嵌套几何体（多边形、多边形、线串）需要递归验证。

标签和元数据净化

标签和其他数字字段可能包含 NaN, Inf，或NumPy数字类型(numpy.float64, numpy.int64)它们不符合JSON标准。所有数字都应该是 转换为原生Python类型，将无效值替换为 None.

NumPy标量有问题

即使是有限的NumPy浮点数或整数也不能直接序列化 json.dumps()，内部使用 JSONResponse() 或 shapely.to_geojson() （最初在代码中用于创建功能）。转换为Python float 或 int 是必需的。

MCP上下文处理

当通过REST调用时， ctx: Context 可能是 None日志记录或其他依赖上下文的操作应受到保护，或使用虚拟上下文以避免 运行错误.

安全递归序列化

使用a 递归净化函数确保几何图形、标签和所有嵌套结构都是完全JSON安全的这保证了MCP端点、REST API、Postman和浏览器测试的兼容性。

并发服务器设置

在同一Python进程中同时运行MCP和REST服务器是可行的 asyncio.create_task 和 asyncio.gather，同时启用代理工作流和HTTP测试。

未决问题（即将发布…）

1.数据检索和处理

  • []使用参数在OverpassQuery中筛选所需列
  • []在构建OverpassFeature时对迭代步骤进行矢量化
  • []使毕旦场适应矢量化方法
  • []实现字典解包作为构建OverpassFeature的替代方案
  • []为OverpassFeature地理数据帧的筛选步骤中使用的字段定义硬类型枚举类型
  • []进一步抽象/基类MCP+工具使用

2.测试基础设施

  • []设置内部Python代码测试套件进行代码执行测试
  • []实现检查循环，在执行之前在测试套件中测试生成的代码
  • []单元测试

3.LLM链实施

  • []OpenAI+Anthropic API聊天端点设置
  • []将工具连接到OpenAI LLM
  • []LLM的进一步抽象/基类
  • []使用OverpassFeature工具设置LLM链以生成地理数据帧
  • []为工具执行创建提示模板（生成地理数据帧）+分析步骤（如果用户要求，分析地理数据框）+可视化绘图生成
  • []如果需要分析：让LLM使用推断列的分组/计数生成分析数据帧
  • []让LLM编写代码以生成适当的图表（例如饼图）

4.前端开发

  • []使用Dash/Plotly构建前端
  • []在前端显示结果图表或分析表

相关 MCP