mapbox-mcp-runtime-patterns by mapbox/mapbox-agent-skills
npx skills add https://github.com/mapbox/mapbox-agent-skills --skill mapbox-mcp-runtime-patterns此技能提供了将 Mapbox MCP 服务器集成到 AI 应用程序中的模式,以便在生产环境中使用地理空间功能。
Mapbox MCP Server 是一个模型上下文协议(MCP)服务器,为 AI 智能体提供地理空间工具:
离线工具(Turf.js):
Mapbox API 工具:
实用工具:
关键优势: 无需手动集成多个 API,即可为您的 AI 应用程序赋予地理空间超能力。
在集成之前,请理解工具之间的关键区别,以帮助您的 LLM 正确选择:
直线距离(离线,即时):
distance_tool、bearing_tool、广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
midpoint_tooldistance_tool路线距离(API,考虑交通):
directions_tool、matrix_tooldirections_tool类别/类型搜索:
category_search_toolcategory_search_tool具体地点/地址:
search_and_geocode_tool、reverse_geocode_toolsearch_and_geocode_tool可达区域(可到达的范围):
isochrone_toolisochrone_tool具体路线(如何到达那里):
directions_tooldirections_tool离线工具(免费,即时):
distance_tool、point_in_polygon_tool、area_toolAPI 工具(需要令牌,计入使用量):
directions_tool、category_search_tool、isochrone_tool最佳实践: 尽可能优先使用离线工具,在需要实时数据或路线规划时使用 API 工具。
最简单的集成 - 使用 Mapbox 托管的 MCP 服务器:
https://mcp.mapbox.com/mcp
无需安装。只需在 Authorization 请求头中传递您的 Mapbox 访问令牌。
优势:
认证:
使用基于令牌的认证(程序化访问的标准方式):
Authorization: Bearer your_mapbox_token
注意: 托管服务器也支持 OAuth,但这主要用于交互式流程(编码助手,而非生产应用程序)。
用于自定义部署或开发:
npm install @mapbox/mcp-server
或通过 npx 直接使用:
npx @mapbox/mcp-server
环境设置:
export MAPBOX_ACCESS_TOKEN="your_token_here"
使用场景: 在 Python 中使用类型安全的工具构建 AI 智能体
常见错误: 当将 pydantic-ai 与 OpenAI 一起使用时,正确的导入是
from pydantic_ai.models.openai import OpenAIChatModel。请勿使用OpenAIModel—— 该类在 pydantic-ai 中不存在,会在运行时抛出 ImportError。
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
import requests
import json
import os
class MapboxMCP:
"""通过托管服务器使用 Mapbox MCP。"""
def __init__(self, token: str = None):
self.url = 'https://mcp.mapbox.com/mcp'
self.headers = {'Content-Type': 'application/json'}
# 使用环境变量或参数中的令牌
token = token or os.getenv('MAPBOX_ACCESS_TOKEN')
if token:
self.headers['Authorization'] = f'Bearer {token}'
def call_tool(self, tool_name: str, params: dict) -> dict:
"""通过 HTTPS 调用 MCP 工具。"""
request = {
'jsonrpc': '2.0',
'id': 1,
'method': 'tools/call',
'params': {
'name': tool_name,
'arguments': params
}
}
response = requests.post(
self.url,
headers=self.headers,
json=request
)
response.raise_for_status()
data = response.json()
if 'error' in data:
raise RuntimeError(f"MCP error: {data['error']['message']}")
return data['result']['content'][0]['text']
# 使用 Mapbox 工具创建智能体
# 直接传递令牌或设置 MAPBOX_ACCESS_TOKEN 环境变量
mapbox = MapboxMCP(token='your_token')
agent = Agent(
model=OpenAIChatModel('gateway/openai:gpt-5.2'),
tools=[
lambda from_loc, to_loc: mapbox.call_tool(
'directions_tool',
{'coordinates': [from_loc, to_loc], 'routing_profile': 'mapbox/driving-traffic'}
),
lambda address: mapbox.call_tool(
'reverse_geocode_tool',
{'coordinates': {'longitude': address[0], 'latitude': address[1]}}
)
]
)
# 使用智能体
result = agent.run_sync(
"What's the driving time from Boston to NYC?"
)
import subprocess
class MapboxMCPLocal:
def __init__(self, token: str):
self.token = token
self.mcp_process = subprocess.Popen(
['npx', '@mapbox/mcp-server'],
env={'MAPBOX_ACCESS_TOKEN': token},
stdin=subprocess.PIPE,
stdout=subprocess.PIPE
)
def call_tool(self, tool_name: str, params: dict) -> dict:
# ... 与托管服务器类似,但通过子进程
pass
优势:
使用场景: 具有地理空间能力的多智能体编排
CrewAI 支持构建具有专门角色的自主智能体团队。与 Mapbox MCP 集成可为您的团队添加地理空间智能。
from crewai import Agent, Task, Crew
from crewai.tools import BaseTool
import requests
import os
from typing import Type
from pydantic import BaseModel, Field
class MapboxMCP:
"""Mapbox MCP 连接器。"""
def __init__(self, token: str = None):
self.url = 'https://mcp.mapbox.com/mcp'
token = token or os.getenv('MAPBOX_ACCESS_TOKEN')
self.headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {token}'
}
def call_tool(self, tool_name: str, params: dict) -> str:
request = {
'jsonrpc': '2.0',
'id': 1,
'method': 'tools/call',
'params': {'name': tool_name, 'arguments': params}
}
response = requests.post(self.url, headers=self.headers, json=request)
response.raise_for_status()
data = response.json()
if 'error' in data:
raise RuntimeError(f"MCP error: {data['error']['message']}")
return data['result']['content'][0]['text']
# 为 CrewAI 创建 Mapbox 工具
class DirectionsTool(BaseTool):
name: str = "directions_tool"
description: str = "获取两个地点之间的驾驶路线"
class InputSchema(BaseModel):
origin: list = Field(description="起点 [经度, 纬度]")
destination: list = Field(description="目的地 [经度, 纬度]")
args_schema: Type[BaseModel] = InputSchema
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def _run(self, origin: list, destination: list) -> str:
result = self.mcp.call_tool('directions_tool', {
'coordinates': [
{'longitude': origin[0], 'latitude': origin[1]},
{'longitude': destination[0], 'latitude': destination[1]}
],
'routing_profile': 'mapbox/driving-traffic'
})
return f"Directions: {result}"
class GeocodeTool(BaseTool):
name: str = "reverse_geocode_tool"
description: str = "将坐标转换为人类可读的地址"
class InputSchema(BaseModel):
coordinates: list = Field(description="坐标 [经度, 纬度]")
args_schema: Type[BaseModel] = InputSchema
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def _run(self, coordinates: list) -> str:
result = self.mcp.call_tool('reverse_geocode_tool', {
'coordinates': {'longitude': coordinates[0], 'latitude': coordinates[1]}
})
return result
class SearchPOITool(BaseTool):
name: str = "search_poi"
description: str = "按类别查找某个地点附近的兴趣点"
class InputSchema(BaseModel):
category: str = Field(description="POI 类别(餐厅、酒店等)")
location: list = Field(description="搜索中心点 [经度, 纬度]")
args_schema: Type[BaseModel] = InputSchema
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def _run(self, category: str, location: list) -> str:
result = self.mcp.call_tool('category_search_tool', {
'category': category,
'proximity': {'longitude': location[0], 'latitude': location[1]}
})
return result
# 使用地理空间工具创建专门的智能体
location_analyst = Agent(
role='位置分析师',
goal='分析地理位置并提供见解',
backstory="""地理分析和位置情报专家。
使用 search_poi 查找地点类型(餐厅、酒店)。
使用 reverse_geocode_tool 将坐标转换为地址。""",
tools=[GeocodeTool(), SearchPOITool()],
verbose=True
)
route_planner = Agent(
role='路线规划师',
goal='规划最优路线并提供行程时间估算',
backstory="""经验丰富的物流协调员,专攻路线优化。
使用 directions_tool 获取考虑交通的沿道路路线距离。
当需要了解交通状况的行程时间时,始终使用此工具。""",
tools=[DirectionsTool()],
verbose=True
)
# 创建任务
find_restaurants_task = Task(
description="""
查找坐标 [-73.9857, 40.7484](时代广场)附近排名前 5 的餐厅。
提供它们的名称和大致距离。
""",
agent=location_analyst,
expected_output="包含距离的 5 家餐厅列表"
)
plan_route_task = Task(
description="""
规划从 [-74.0060, 40.7128](纽约市中心)到 [-73.9857, 40.7484](时代广场)的路线。
考虑当前交通状况,提供驾驶时间。
""",
agent=route_planner,
expected_output="包含预计驾驶时间的路线"
)
# 创建并运行团队
crew = Crew(
agents=[location_analyst, route_planner],
tasks=[find_restaurants_task, plan_route_task],
verbose=True
)
result = crew.kickoff()
print(result)
真实世界示例 - 餐厅查找团队:
# 为餐厅推荐系统定义团队
class RestaurantCrew:
def __init__(self):
self.mcp = MapboxMCP()
# 位置专家智能体
self.location_agent = Agent(
role='位置专家',
goal='查找并分析餐厅位置',
tools=[SearchPOITool(), GeocodeTool()],
backstory='寻找最佳就餐地点的专家'
)
# 物流智能体
self.logistics_agent = Agent(
role='物流协调员',
goal='计算行程时间和最优路线',
tools=[DirectionsTool()],
backstory='城市导航和时间优化专家'
)
def find_restaurants_with_commute(self, user_location: list, max_minutes: int):
# 任务 1:查找附近餐厅
search_task = Task(
description=f"查找 {user_location} 附近的餐厅",
agent=self.location_agent,
expected_output="包含坐标的餐厅列表"
)
# 任务 2:计算行程时间
route_task = Task(
description=f"计算从 {user_location} 到每个餐厅的行程时间",
agent=self.logistics_agent,
expected_output="到每个餐厅的行程时间",
context=[search_task] # 依赖于搜索结果
)
crew = Crew(
agents=[self.location_agent, self.logistics_agent],
tasks=[search_task, route_task],
verbose=True
)
return crew.kickoff()
# 使用
restaurant_crew = RestaurantCrew()
results = restaurant_crew.find_restaurants_with_commute(
user_location=[-73.9857, 40.7484],
max_minutes=15
)
优势:
使用场景: 具有地理空间能力的轻量级智能体(Hugging Face)
Smolagents 是 Hugging Face 的简单、高效的智能体框架。非常适合以最小开销部署地理空间智能体。
from smolagents import CodeAgent, Tool, HfApiModel
import requests
import os
class MapboxMCP:
"""Mapbox MCP 连接器。"""
def __init__(self, token: str = None):
self.url = 'https://mcp.mapbox.com/mcp'
token = token or os.getenv('MAPBOX_ACCESS_TOKEN')
self.headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {token}'
}
def call_tool(self, tool_name: str, params: dict) -> str:
request = {
'jsonrpc': '2.0',
'id': 1,
'method': 'tools/call',
'params': {'name': tool_name, 'arguments': params}
}
response = requests.post(self.url, headers=self.headers, json=request)
result = response.json()['result']
return result['content'][0]['text']
# 为 Smolagents 创建 Mapbox 工具
class DirectionsTool(Tool):
name = "directions_tool"
description = """
获取两个地点之间的驾驶路线。
参数:
origin: 起点坐标 [经度, 纬度]
destination: 目的地坐标 [经度, 纬度]
返回:
包含距离和行程时间的路线
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, origin: list, destination: list) -> str:
return self.mcp.call_tool('directions_tool', {
'coordinates': [
{'longitude': origin[0], 'latitude': origin[1]},
{'longitude': destination[0], 'latitude': destination[1]}
],
'routing_profile': 'mapbox/driving-traffic'
})
class CalculateDistanceTool(Tool):
name = "distance_tool"
description = """
计算两点之间的距离(离线,即时)。
参数:
from_coords: 起点坐标 [经度, 纬度]
to_coords: 终点坐标 [经度, 纬度]
units: 'miles' 或 'kilometers'
返回:
距离数值
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, from_coords: list, to_coords: list, units: str = 'miles') -> str:
return self.mcp.call_tool('distance_tool', {
'from': {'longitude': from_coords[0], 'latitude': from_coords[1]},
'to': {'longitude': to_coords[0], 'latitude': to_coords[1]},
'units': units
})
class SearchPOITool(Tool):
name = "search_poi"
description = """
按类别搜索兴趣点。
参数:
category: POI 类别(餐厅、酒店、加油站等)
location: 搜索中心点 [经度, 纬度]
返回:
附近 POI 的列表,包含名称和坐标
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, category: str, location: list) -> str:
return self.mcp.call_tool('category_search_tool', {
'category': category,
'proximity': {'longitude': location[0], 'latitude': location[1]}
})
class IsochroneTool(Tool):
name = "isochrone_tool"
description = """
计算时间限制内的可达区域(等时线)。
参数:
location: 中心点 [经度, 纬度]
minutes: 时间限制(分钟)
profile: 'mapbox/driving'、'mapbox/walking' 或 'mapbox/cycling'
返回:
可达区域的 GeoJSON 多边形
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, location: list, minutes: int, profile: str = 'mapbox/driving') -> str:
return self.mcp.call_tool('isochrone_tool', {
'coordinates': {'longitude': location[0], 'latitude': location[1]},
'contours_minutes': [minutes],
'profile': profile
})
# 使用 Mapbox 工具创建智能体
model = HfApiModel()
agent = CodeAgent(
tools=[
DirectionsTool(),
CalculateDistanceTool(),
SearchPOITool(),
IsochroneTool()
],
model=model
)
# 使用智能体
result = agent.run(
"Find restaurants within 10 minutes walking from Times Square NYC "
"(coordinates: -73.9857, 40.7484). Calculate distances to each."
)
print(result)
真实世界示例 - 房产搜索智能体:
class PropertySearchAgent:
def __init__(self):
self.mcp = MapboxMCP()
# 创建专门的工具
tools = [
IsochroneTool(),
SearchPOITool(),
CalculateDistanceTool()
]
self.agent = CodeAgent(
tools=tools,
model=HfApiModel()
)
def find_properties_near_work(
self,
work_location: list,
max_commute_minutes: int,
property_locations: list[dict]
):
"""查找工作地点通勤时间内的房产。"""
prompt = f"""
I need to find properties within {max_commute_minutes} minutes
driving of my work at {work_location}.
Property locations to check:
{property_locations}
For each property:
1. Calculate if it's within the commute time
2. Find nearby amenities (grocery stores, restaurants)
3. Calculate distances to key locations
Return a ranked list of properties with commute time and nearby amenities.
"""
return self.agent.run(prompt)
# 使用
property_agent = PropertySearchAgent()
properties = [
{'id': 1, 'address': '123 Main St', 'coords': [-122.4194, 37.7749]},
{'id': 2, 'address': '456 Oak Ave', 'coords': [-122.4094, 37.7849]},
]
results = property_agent.find_properties_near_work(
work_location=[-122.4, 37.79], # 旧金山市中心
max_commute_minutes=30,
property_locations=properties
)
优势:
使用场景: 构建具有地理空间工作流的多智能体系统
import { Mastra } from '@mastra/core';
class MapboxMCP {
private url = 'https://mcp.mapbox.com/mcp';
private headers: Record<string, string>;
constructor(token?: string) {
const mapboxToken = token || process.env.MAPBOX_ACCESS_TOKEN;
this.headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${mapboxToken}`
};
}
async callTool(toolName: string, params: any): Promise<any> {
const request = {
jsonrpc: '2.0',
id: Date.now(),
method: 'tools/call',
params: { name: toolName, arguments: params }
};
const response = await fetch(this.url, {
method: 'POST',
headers: this.headers,
body: JSON.stringify(request)
});
const data = await response.json();
return JSON.parse(data.result.content[0].text);
}
}
// 使用 Mapbox 工具创建 Mastra 智能体
import { Agent } from '@mastra/core/agent';
import { createTool } from '@mastra/core/tools';
import { z } from 'zod';
const mcp = new MapboxMCP();
// 创建 Mapbox 工具
const searchPOITool = createTool({
id: 'search-poi',
description: '查找特定类别的地点附近的位置',
inputSchema: z.object({
category: z.string(),
location: z.array(z.number()).length(2)
}),
execute: async ({ category, location }) => {
return await mcp.callTool('category_search_tool', {
category,
proximity: { longitude: location[0], latitude: location[1] }
});
}
});
const getDirectionsTool = createTool({
id: 'get-directions',
description: '获取考虑交通的驾驶路线',
inputSchema: z.object({
origin: z.array(z.number()).length(2),
destination: z.array(z.number()).length(2)
}),
execute: async ({ origin, destination }) => {
return await mcp.callTool('directions_tool', {
coordinates: [
{ longitude: origin[0], latitude: origin[1] },
{ longitude: destination[0], latitude: destination[1] }
],
routing_profile: 'mapbox/driving-traffic'
});
}
});
// 创建位置智能体
const locationAgent = new Agent({
id: 'location-agent',
name: '位置情报智能体',
instructions: '您帮助用户使用地理空间工具查找地点和规划路线。',
model: 'openai/gpt-5.2',
tools: {
searchPOITool,
getDirectionsTool
}
});
// 使用智能体
const result = await locationAgent.generate([
{ role: 'user', content: 'Find restaurants near Times Square NYC (-73.9857, 40.7484)' }
]);
优势:
使用场景: 构建具有地理空间工具的对话式 AI
import { ChatOpenAI } from '@langchain/openai';
import { AgentExecutor, createToolCallingAgent } from 'langchain/agents';
import { DynamicStructuredTool } from '@langchain/core/tools';
import { ChatPromptTemplate, MessagesPlaceholder } from '@langchain/core/prompts';
import { z } from 'zod';
// 托管服务器的 MCP 服务器包装器
class MapboxMCP {
private url = 'https://mcp.mapbox.com/mcp';
private headers: Record<string, string>;
constructor(token?: string) {
const mapboxToken = token || process.env.MAPBOX_ACCESS_TOKEN;
this.headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${mapboxToken}`
};
}
async callTool(name: string, args: any): Promise<string> {
const request = {
jsonrpc: '2.0',
id: Date.now(),
method: 'tools/call',
params: { name, arguments: args }
};
const response = await fetch(this.url, {
method: 'POST',
headers: this.headers,
body: JSON.stringify(request)
});
const data = await response.json();
return data.result.content[0].text;
}
}
// 从 MCP 创建 LangChain 工具
const mcp = new MapboxMCP();
const tools = [
new DynamicStructuredTool({
name: 'directions_tool',
description:
'获取考虑交通状况的逐向驾驶路线。当您需要实际驾驶路线或考虑交通状况的持续时间时使用。',
schema: z.object({
origin: z.tuple([z.number(), z.number()]).describe('起点 [经度, 纬度]'),
destination: z.tuple([z.number(), z.number()]).describe('目的地 [经度, 纬度]')
}) as any,
func: async ({ origin, destination }: any) => {
return await mcp.callTool('directions_tool', {
coordinates: [
{ longitude: origin[0], latitude: origin[1] },
{ longitude: destination[0], latitude: destination[1] }
],
routing_profile: 'mapbox/driving-traffic'
});
}
}),
new DynamicStructuredTool({
name: 'category_search_tool',
description:
'查找特定类别类型的所有地点附近的位置。当用户想要按类型(餐厅、酒店、咖啡店等)浏览地点时使用。',
schema: z.object({
category: z.string().describe('POI 类别:餐厅、酒店、咖啡店等。'),
location: z.tuple([z.number(), z.number()]).describe('搜索中心点 [经度, 纬度]')
}) as any,
func: async ({ category, location }: any) => {
return await mcp.callTool('category_search_tool', {
category,
proximity: { longitude: location[0], latitude: location[1] }
});
}
}),
new DynamicStructuredTool({
name: 'isochrone_tool',
description:
'计算从起点出发,在时间限制内可到达的区域。用于回答"我在 X 分钟内可以到达哪里?"这类问题。',
schema: z.object({
location: z.tuple([z.number(), z.number()]).describe('中心点 [经度, 纬度]'),
minutes: z.number().describe('时间限制(分钟)'),
profile: z.enum(['mapbox/driving', 'mapbox/walking', 'mapbox/cycling']).optional()
}) as any,
func: async ({ location, minutes, profile }: any) => {
return await mcp.callTool('isochrone_tool', {
coordinates: { longitude: location[0], latitude: location[1] },
contours_minutes: [minutes],
profile: profile || 'mapbox/walking'
});
}
}),
new DynamicStructuredTool({
name: 'distance_tool',
description: '计算两点之间的直线距离(离线,免费)',
schema: z.object({
from: z.tuple([z.number(), z.number()]).describe('起点 [经度, 纬度]'),
to: z.tuple([z.number(), z.number()]).describe('终点 [经度, 纬度]'),
units: z.enum(['miles', 'kilometers']).optional()
}) as any,
func: async ({ from, to, units }: any) => {
return await mcp.callTool('distance_tool', {
from: { longitude: from[0], latitude: from[1] },
to: { longitude: to[0], latitude: to[1] },
units: units || 'miles'
});
}
})
];
// 创建智能体
const llm = new ChatOpenAI({ model: 'gpt-5.2', temperature: 0 });
const prompt = ChatPromptTemplate.fromMessages([
['system', '您是一个位置情报助手。'],
['human', '{input}'],
new MessagesPlaceholder('agent_scratchpad')
]);
// @ts-ignore - Zod 元组模式会导致深层类型递归
const agent = await createToolCallingAgent({ llm, tools, prompt });
const executor = new AgentExecutor({ agent, tools, verbose: true });
// 使用智能体
const result = await executor.invoke({
input: 'Find coffee shops within 10 minutes walking from Union Square, NYC'
});
优势:
TypeScript 类型注意事项:
当将 DynamicStructuredTool 与 Zod 模式(特别是 z.tuple())一起使用时,TypeScript 可能会遇到深层类型递归错误。这是复杂 Zod 泛型类型的已知限制。最小的修复方法是添加 as any 类型断言:
const tool = new DynamicStructuredTool({
name: 'my_tool',
schema: z.object({
coords: z.tuple([z.number(), z.number()])
}) as any, // ← 添加 'as any' 以防止类型递归
func: async ({ coords }: any) => {
// ← 将参数类型设为 'any'
// 实现
}
});
// 对于来自外部 API 的 JSON 响应
const data = (await response.json()) as any;
// 对于具有复杂工具类型的 createOpenAIFunctionsAgent
// @ts-ignore - Zod 元组模式会导致深层类型递归
const agent = await createOpenAIFunctionsAgent({ llm, tools, prompt });
这不会影响运行时验证(Zod 仍在运行时验证)—— 它只帮助 TypeScript 的类型检查器在编译期间避免无限递归。
使用场景: 构建领域特定的 AI 应用程序(Zillow 风格、TripAdvisor 风格)
interface MCPTool {
name: string;
description: string;
inputSchema: any;
}
class CustomMapboxAgent {
private url = 'https://mcp.mapbox.com/mcp';
private headers: Record<string, string>;
private tools: Map<string, MCPTool> = new Map();
constructor(token?: string) {
const mapboxToken = token || process.env.MAPBOX_ACCESS_TOKEN;
this.headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${mapboxToken}`
};
}
async initialize() {
// 从 MCP 服务器发现可用工具
await this.discoverTools();
}
private async discoverTools() {
const request = {
jsonrpc: '2.0',
id: 1,
method: 'tools/list'
};
const response = await this.sendMCPRequest(request);
response.result.tools.forEach((tool: MCPTool) => {
this.tools.set(tool.name, tool);
});
}
async callTool(toolName: string, params: any): Promise<any> {
const request = {
jsonrpc: '2.0',
id: Date.now(),
method: 'tools/call',
params: { name: toolName, arguments: params }
};
const response = await this.sendMCPRequest(request);
return response.result.content[0].text;
}
private async sendMCPRequest(request: any): Promise<any> {
const response = await fetch(this.url, {
method: 'POST',
headers: this.headers,
body: JSON.stringify(request)
});
const data = await response.json();
if (data.error) {
throw new Error(data.error.message);
}
return data;
}
// 领域特定的方法
async findPropertiesWithCommute(
homeLocation: [number, number],
workLocation: [number, number],
maxCommuteMinutes: number
) {
// 从工作地点获取等时线
const isochrone = await this.callTool('isochrone_tool', {
coordinates: { longitude: workLocation[0], latitude: workLocation[1] },
contours_minutes: [maxCommuteMinutes],
profile: 'mapbox/driving-traffic'
});
// 检查家庭住址是否在等时线内
const isInRange = await this.callTool('point_in_polygon_tool', {
point: { longitude: homeLocation[0], latitude: homeLocation[1] },
polygon: JSON.parse(isochrone).features[0].geometry
});
return JSON.parse(isInRange);
}
async findRestaurantsNearby(location: [number, number], radiusMiles: number) {
// 搜索餐厅
const results = await this.callTool('category_search_tool', {
category: 'restaurant',
proximity: { longitude: location[0], latitude: location[1] }
});
// 按距离过滤
const restaurants =
This skill provides patterns for integrating the Mapbox MCP Server into AI applications for production use with geospatial capabilities.
The Mapbox MCP Server is a Model Context Protocol (MCP) server that provides AI agents with geospatial tools:
Offline Tools (Turf.js):
Mapbox API Tools:
Utility Tools:
Key benefit: Give your AI application geospatial superpowers without manually integrating multiple APIs.
Before integrating, understand the key distinctions between tools to help your LLM choose correctly:
Straight-line distance (offline, instant):
distance_tool, bearing_tool, midpoint_tooldistance_toolRoute distance (API, traffic-aware):
directions_tool, matrix_tooldirections_toolCategory/type search :
category_search_toolcategory_search_toolSpecific place/address :
search_and_geocode_tool, reverse_geocode_toolsearch_and_geocode_toolReachable area (what's within reach):
isochrone_toolisochrone_toolSpecific route (how to get there):
directions_tooldirections_toolOffline tools (free, instant):
distance_tool, point_in_polygon_tool, area_toolAPI tools (requires token, counts against usage):
directions_tool, category_search_tool, isochrone_toolBest practice: Prefer offline tools when possible, use API tools when you need real-time data or routing.
Easiest integration - Use Mapbox's hosted MCP server at:
https://mcp.mapbox.com/mcp
No installation required. Simply pass your Mapbox access token in the Authorization header.
Benefits:
Authentication:
Use token-based authentication (standard for programmatic access):
Authorization: Bearer your_mapbox_token
Note: The hosted server also supports OAuth, but that's primarily for interactive flows (coding assistants, not production apps).
For custom deployments or development:
npm install @mapbox/mcp-server
Or use directly via npx:
npx @mapbox/mcp-server
Environment setup:
export MAPBOX_ACCESS_TOKEN="your_token_here"
Use case: Building AI agents with type-safe tools in Python
Common mistake: When using pydantic-ai with OpenAI, the correct import is
from pydantic_ai.models.openai import OpenAIChatModel. Do NOT useOpenAIModel— that class does not exist in pydantic-ai and will throw an ImportError at runtime.
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
import requests
import json
import os
class MapboxMCP:
"""Mapbox MCP via hosted server."""
def __init__(self, token: str = None):
self.url = 'https://mcp.mapbox.com/mcp'
self.headers = {'Content-Type': 'application/json'}
# Use token from environment or parameter
token = token or os.getenv('MAPBOX_ACCESS_TOKEN')
if token:
self.headers['Authorization'] = f'Bearer {token}'
def call_tool(self, tool_name: str, params: dict) -> dict:
"""Call MCP tool via HTTPS."""
request = {
'jsonrpc': '2.0',
'id': 1,
'method': 'tools/call',
'params': {
'name': tool_name,
'arguments': params
}
}
response = requests.post(
self.url,
headers=self.headers,
json=request
)
response.raise_for_status()
data = response.json()
if 'error' in data:
raise RuntimeError(f"MCP error: {data['error']['message']}")
return data['result']['content'][0]['text']
# Create agent with Mapbox tools
# Pass token directly or set MAPBOX_ACCESS_TOKEN env var
mapbox = MapboxMCP(token='your_token')
agent = Agent(
model=OpenAIChatModel('gateway/openai:gpt-5.2'),
tools=[
lambda from_loc, to_loc: mapbox.call_tool(
'directions_tool',
{'coordinates': [from_loc, to_loc], 'routing_profile': 'mapbox/driving-traffic'}
),
lambda address: mapbox.call_tool(
'reverse_geocode_tool',
{'coordinates': {'longitude': address[0], 'latitude': address[1]}}
)
]
)
# Use agent
result = agent.run_sync(
"What's the driving time from Boston to NYC?"
)
import subprocess
class MapboxMCPLocal:
def __init__(self, token: str):
self.token = token
self.mcp_process = subprocess.Popen(
['npx', '@mapbox/mcp-server'],
env={'MAPBOX_ACCESS_TOKEN': token},
stdin=subprocess.PIPE,
stdout=subprocess.PIPE
)
def call_tool(self, tool_name: str, params: dict) -> dict:
# ... similar to hosted but via subprocess
pass
Benefits:
Use case: Multi-agent orchestration with geospatial capabilities
CrewAI enables building autonomous agent crews with specialized roles. Integration with Mapbox MCP adds geospatial intelligence to your crew.
from crewai import Agent, Task, Crew
from crewai.tools import BaseTool
import requests
import os
from typing import Type
from pydantic import BaseModel, Field
class MapboxMCP:
"""Mapbox MCP connector."""
def __init__(self, token: str = None):
self.url = 'https://mcp.mapbox.com/mcp'
token = token or os.getenv('MAPBOX_ACCESS_TOKEN')
self.headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {token}'
}
def call_tool(self, tool_name: str, params: dict) -> str:
request = {
'jsonrpc': '2.0',
'id': 1,
'method': 'tools/call',
'params': {'name': tool_name, 'arguments': params}
}
response = requests.post(self.url, headers=self.headers, json=request)
response.raise_for_status()
data = response.json()
if 'error' in data:
raise RuntimeError(f"MCP error: {data['error']['message']}")
return data['result']['content'][0]['text']
# Create Mapbox tools for CrewAI
class DirectionsTool(BaseTool):
name: str = "directions_tool"
description: str = "Get driving directions between two locations"
class InputSchema(BaseModel):
origin: list = Field(description="Origin [lng, lat]")
destination: list = Field(description="Destination [lng, lat]")
args_schema: Type[BaseModel] = InputSchema
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def _run(self, origin: list, destination: list) -> str:
result = self.mcp.call_tool('directions_tool', {
'coordinates': [
{'longitude': origin[0], 'latitude': origin[1]},
{'longitude': destination[0], 'latitude': destination[1]}
],
'routing_profile': 'mapbox/driving-traffic'
})
return f"Directions: {result}"
class GeocodeTool(BaseTool):
name: str = "reverse_geocode_tool"
description: str = "Convert coordinates to human-readable address"
class InputSchema(BaseModel):
coordinates: list = Field(description="Coordinates [lng, lat]")
args_schema: Type[BaseModel] = InputSchema
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def _run(self, coordinates: list) -> str:
result = self.mcp.call_tool('reverse_geocode_tool', {
'coordinates': {'longitude': coordinates[0], 'latitude': coordinates[1]}
})
return result
class SearchPOITool(BaseTool):
name: str = "search_poi"
description: str = "Find points of interest by category near a location"
class InputSchema(BaseModel):
category: str = Field(description="POI category (restaurant, hotel, etc.)")
location: list = Field(description="Search center [lng, lat]")
args_schema: Type[BaseModel] = InputSchema
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def _run(self, category: str, location: list) -> str:
result = self.mcp.call_tool('category_search_tool', {
'category': category,
'proximity': {'longitude': location[0], 'latitude': location[1]}
})
return result
# Create specialized agents with geospatial tools
location_analyst = Agent(
role='Location Analyst',
goal='Analyze geographic locations and provide insights',
backstory="""Expert in geographic analysis and location intelligence.
Use search_poi for finding types of places (restaurants, hotels).
Use reverse_geocode_tool for converting coordinates to addresses.""",
tools=[GeocodeTool(), SearchPOITool()],
verbose=True
)
route_planner = Agent(
role='Route Planner',
goal='Plan optimal routes and provide travel time estimates',
backstory="""Experienced logistics coordinator specializing in route optimization.
Use directions_tool for route distance along roads with traffic.
Always use when traffic-aware travel time is needed.""",
tools=[DirectionsTool()],
verbose=True
)
# Create tasks
find_restaurants_task = Task(
description="""
Find the top 5 restaurants near coordinates [-73.9857, 40.7484] (Times Square).
Provide their names and approximate distances.
""",
agent=location_analyst,
expected_output="List of 5 restaurants with distances"
)
plan_route_task = Task(
description="""
Plan a route from [-74.0060, 40.7128] (downtown NYC) to [-73.9857, 40.7484] (Times Square).
Provide driving time considering current traffic.
""",
agent=route_planner,
expected_output="Route with estimated driving time"
)
# Create and run crew
crew = Crew(
agents=[location_analyst, route_planner],
tasks=[find_restaurants_task, plan_route_task],
verbose=True
)
result = crew.kickoff()
print(result)
Real-world example - Restaurant finder crew:
# Define crew for restaurant recommendation system
class RestaurantCrew:
def __init__(self):
self.mcp = MapboxMCP()
# Location specialist agent
self.location_agent = Agent(
role='Location Specialist',
goal='Find and analyze restaurant locations',
tools=[SearchPOITool(), GeocodeTool()],
backstory='Expert in finding the best dining locations'
)
# Logistics agent
self.logistics_agent = Agent(
role='Logistics Coordinator',
goal='Calculate travel times and optimal routes',
tools=[DirectionsTool()],
backstory='Specialist in urban navigation and time optimization'
)
def find_restaurants_with_commute(self, user_location: list, max_minutes: int):
# Task 1: Find nearby restaurants
search_task = Task(
description=f"Find restaurants near {user_location}",
agent=self.location_agent,
expected_output="List of restaurants with coordinates"
)
# Task 2: Calculate travel times
route_task = Task(
description=f"Calculate travel time to each restaurant from {user_location}",
agent=self.logistics_agent,
expected_output="Travel times to each restaurant",
context=[search_task] # Depends on search results
)
crew = Crew(
agents=[self.location_agent, self.logistics_agent],
tasks=[search_task, route_task],
verbose=True
)
return crew.kickoff()
# Usage
restaurant_crew = RestaurantCrew()
results = restaurant_crew.find_restaurants_with_commute(
user_location=[-73.9857, 40.7484],
max_minutes=15
)
Benefits:
Use case: Lightweight agents with geospatial capabilities (Hugging Face)
Smolagents is Hugging Face's simple, efficient agent framework. Perfect for deploying geospatial agents with minimal overhead.
from smolagents import CodeAgent, Tool, HfApiModel
import requests
import os
class MapboxMCP:
"""Mapbox MCP connector."""
def __init__(self, token: str = None):
self.url = 'https://mcp.mapbox.com/mcp'
token = token or os.getenv('MAPBOX_ACCESS_TOKEN')
self.headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {token}'
}
def call_tool(self, tool_name: str, params: dict) -> str:
request = {
'jsonrpc': '2.0',
'id': 1,
'method': 'tools/call',
'params': {'name': tool_name, 'arguments': params}
}
response = requests.post(self.url, headers=self.headers, json=request)
result = response.json()['result']
return result['content'][0]['text']
# Create Mapbox tools for Smolagents
class DirectionsTool(Tool):
name = "directions_tool"
description = """
Get driving directions between two locations.
Args:
origin: Origin coordinates as [longitude, latitude]
destination: Destination coordinates as [longitude, latitude]
Returns:
Directions with distance and travel time
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, origin: list, destination: list) -> str:
return self.mcp.call_tool('directions_tool', {
'coordinates': [
{'longitude': origin[0], 'latitude': origin[1]},
{'longitude': destination[0], 'latitude': destination[1]}
],
'routing_profile': 'mapbox/driving-traffic'
})
class CalculateDistanceTool(Tool):
name = "distance_tool"
description = """
Calculate distance between two points (offline, instant).
Args:
from_coords: Start coordinates [longitude, latitude]
to_coords: End coordinates [longitude, latitude]
units: 'miles' or 'kilometers'
Returns:
Distance as a number
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, from_coords: list, to_coords: list, units: str = 'miles') -> str:
return self.mcp.call_tool('distance_tool', {
'from': {'longitude': from_coords[0], 'latitude': from_coords[1]},
'to': {'longitude': to_coords[0], 'latitude': to_coords[1]},
'units': units
})
class SearchPOITool(Tool):
name = "search_poi"
description = """
Search for points of interest by category.
Args:
category: POI category (restaurant, hotel, gas_station, etc.)
location: Search center [longitude, latitude]
Returns:
List of nearby POIs with names and coordinates
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, category: str, location: list) -> str:
return self.mcp.call_tool('category_search_tool', {
'category': category,
'proximity': {'longitude': location[0], 'latitude': location[1]}
})
class IsochroneTool(Tool):
name = "isochrone_tool"
description = """
Calculate reachable area within time limit (isochrone).
Args:
location: Center point [longitude, latitude]
minutes: Time limit in minutes
profile: 'mapbox/driving', 'mapbox/walking', or 'mapbox/cycling'
Returns:
GeoJSON polygon of reachable area
"""
def __init__(self):
super().__init__()
self.mcp = MapboxMCP()
def forward(self, location: list, minutes: int, profile: str = 'mapbox/driving') -> str:
return self.mcp.call_tool('isochrone_tool', {
'coordinates': {'longitude': location[0], 'latitude': location[1]},
'contours_minutes': [minutes],
'profile': profile
})
# Create agent with Mapbox tools
model = HfApiModel()
agent = CodeAgent(
tools=[
DirectionsTool(),
CalculateDistanceTool(),
SearchPOITool(),
IsochroneTool()
],
model=model
)
# Use agent
result = agent.run(
"Find restaurants within 10 minutes walking from Times Square NYC "
"(coordinates: -73.9857, 40.7484). Calculate distances to each."
)
print(result)
Real-world example - Property search agent:
class PropertySearchAgent:
def __init__(self):
self.mcp = MapboxMCP()
# Create specialized tools
tools = [
IsochroneTool(),
SearchPOITool(),
CalculateDistanceTool()
]
self.agent = CodeAgent(
tools=tools,
model=HfApiModel()
)
def find_properties_near_work(
self,
work_location: list,
max_commute_minutes: int,
property_locations: list[dict]
):
"""Find properties within commute time of work."""
prompt = f"""
I need to find properties within {max_commute_minutes} minutes
driving of my work at {work_location}.
Property locations to check:
{property_locations}
For each property:
1. Calculate if it's within the commute time
2. Find nearby amenities (grocery stores, restaurants)
3. Calculate distances to key locations
Return a ranked list of properties with commute time and nearby amenities.
"""
return self.agent.run(prompt)
# Usage
property_agent = PropertySearchAgent()
properties = [
{'id': 1, 'address': '123 Main St', 'coords': [-122.4194, 37.7749]},
{'id': 2, 'address': '456 Oak Ave', 'coords': [-122.4094, 37.7849]},
]
results = property_agent.find_properties_near_work(
work_location=[-122.4, 37.79], # Downtown SF
max_commute_minutes=30,
property_locations=properties
)
Benefits:
Use case: Building multi-agent systems with geospatial workflows
import { Mastra } from '@mastra/core';
class MapboxMCP {
private url = 'https://mcp.mapbox.com/mcp';
private headers: Record<string, string>;
constructor(token?: string) {
const mapboxToken = token || process.env.MAPBOX_ACCESS_TOKEN;
this.headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${mapboxToken}`
};
}
async callTool(toolName: string, params: any): Promise<any> {
const request = {
jsonrpc: '2.0',
id: Date.now(),
method: 'tools/call',
params: { name: toolName, arguments: params }
};
const response = await fetch(this.url, {
method: 'POST',
headers: this.headers,
body: JSON.stringify(request)
});
const data = await response.json();
return JSON.parse(data.result.content[0].text);
}
}
// Create Mastra agent with Mapbox tools
import { Agent } from '@mastra/core/agent';
import { createTool } from '@mastra/core/tools';
import { z } from 'zod';
const mcp = new MapboxMCP();
// Create Mapbox tools
const searchPOITool = createTool({
id: 'search-poi',
description: 'Find places of a specific category near a location',
inputSchema: z.object({
category: z.string(),
location: z.array(z.number()).length(2)
}),
execute: async ({ category, location }) => {
return await mcp.callTool('category_search_tool', {
category,
proximity: { longitude: location[0], latitude: location[1] }
});
}
});
const getDirectionsTool = createTool({
id: 'get-directions',
description: 'Get driving directions with traffic',
inputSchema: z.object({
origin: z.array(z.number()).length(2),
destination: z.array(z.number()).length(2)
}),
execute: async ({ origin, destination }) => {
return await mcp.callTool('directions_tool', {
coordinates: [
{ longitude: origin[0], latitude: origin[1] },
{ longitude: destination[0], latitude: destination[1] }
],
routing_profile: 'mapbox/driving-traffic'
});
}
});
// Create location agent
const locationAgent = new Agent({
id: 'location-agent',
name: 'Location Intelligence Agent',
instructions: 'You help users find places and plan routes with geospatial tools.',
model: 'openai/gpt-5.2',
tools: {
searchPOITool,
getDirectionsTool
}
});
// Use agent
const result = await locationAgent.generate([
{ role: 'user', content: 'Find restaurants near Times Square NYC (-73.9857, 40.7484)' }
]);
Benefits:
Use case: Building conversational AI with geospatial tools
import { ChatOpenAI } from '@langchain/openai';
import { AgentExecutor, createToolCallingAgent } from 'langchain/agents';
import { DynamicStructuredTool } from '@langchain/core/tools';
import { ChatPromptTemplate, MessagesPlaceholder } from '@langchain/core/prompts';
import { z } from 'zod';
// MCP Server wrapper for hosted server
class MapboxMCP {
private url = 'https://mcp.mapbox.com/mcp';
private headers: Record<string, string>;
constructor(token?: string) {
const mapboxToken = token || process.env.MAPBOX_ACCESS_TOKEN;
this.headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${mapboxToken}`
};
}
async callTool(name: string, args: any): Promise<string> {
const request = {
jsonrpc: '2.0',
id: Date.now(),
method: 'tools/call',
params: { name, arguments: args }
};
const response = await fetch(this.url, {
method: 'POST',
headers: this.headers,
body: JSON.stringify(request)
});
const data = await response.json();
return data.result.content[0].text;
}
}
// Create LangChain tools from MCP
const mcp = new MapboxMCP();
const tools = [
new DynamicStructuredTool({
name: 'directions_tool',
description:
'Get turn-by-turn driving directions with traffic-aware route distance along roads. Use when you need the actual driving route or traffic-aware duration.',
schema: z.object({
origin: z.tuple([z.number(), z.number()]).describe('Origin [longitude, latitude]'),
destination: z.tuple([z.number(), z.number()]).describe('Destination [longitude, latitude]')
}) as any,
func: async ({ origin, destination }: any) => {
return await mcp.callTool('directions_tool', {
coordinates: [
{ longitude: origin[0], latitude: origin[1] },
{ longitude: destination[0], latitude: destination[1] }
],
routing_profile: 'mapbox/driving-traffic'
});
}
}),
new DynamicStructuredTool({
name: 'category_search_tool',
description:
'Find ALL places of a specific category type near a location. Use when user wants to browse places by type (restaurants, hotels, coffee, etc.).',
schema: z.object({
category: z.string().describe('POI category: restaurant, hotel, coffee, etc.'),
location: z.tuple([z.number(), z.number()]).describe('Search center [longitude, latitude]')
}) as any,
func: async ({ category, location }: any) => {
return await mcp.callTool('category_search_tool', {
category,
proximity: { longitude: location[0], latitude: location[1] }
});
}
}),
new DynamicStructuredTool({
name: 'isochrone_tool',
description:
'Calculate the AREA reachable within a time limit from a starting point. Use for "What can I reach in X minutes?" questions.',
schema: z.object({
location: z.tuple([z.number(), z.number()]).describe('Center point [longitude, latitude]'),
minutes: z.number().describe('Time limit in minutes'),
profile: z.enum(['mapbox/driving', 'mapbox/walking', 'mapbox/cycling']).optional()
}) as any,
func: async ({ location, minutes, profile }: any) => {
return await mcp.callTool('isochrone_tool', {
coordinates: { longitude: location[0], latitude: location[1] },
contours_minutes: [minutes],
profile: profile || 'mapbox/walking'
});
}
}),
new DynamicStructuredTool({
name: 'distance_tool',
description: 'Calculate straight-line distance between two points (offline, free)',
schema: z.object({
from: z.tuple([z.number(), z.number()]).describe('Start [longitude, latitude]'),
to: z.tuple([z.number(), z.number()]).describe('End [longitude, latitude]'),
units: z.enum(['miles', 'kilometers']).optional()
}) as any,
func: async ({ from, to, units }: any) => {
return await mcp.callTool('distance_tool', {
from: { longitude: from[0], latitude: from[1] },
to: { longitude: to[0], latitude: to[1] },
units: units || 'miles'
});
}
})
];
// Create agent
const llm = new ChatOpenAI({ model: 'gpt-5.2', temperature: 0 });
const prompt = ChatPromptTemplate.fromMessages([
['system', 'You are a location intelligence assistant.'],
['human', '{input}'],
new MessagesPlaceholder('agent_scratchpad')
]);
// @ts-ignore - Zod tuple schemas cause deep type recursion
const agent = await createToolCallingAgent({ llm, tools, prompt });
const executor = new AgentExecutor({ agent, tools, verbose: true });
// Use agent
const result = await executor.invoke({
input: 'Find coffee shops within 10 minutes walking from Union Square, NYC'
});
Benefits:
TypeScript Type Considerations:
When using DynamicStructuredTool with Zod schemas (especially z.tuple()), TypeScript may encounter deep type recursion errors. This is a known limitation with complex Zod generic types. The minimal fix is to add as any type assertions:
const tool = new DynamicStructuredTool({
name: 'my_tool',
schema: z.object({
coords: z.tuple([z.number(), z.number()])
}) as any, // ← Add 'as any' to prevent type recursion
func: async ({ coords }: any) => {
// ← Type parameters as 'any'
// Implementation
}
});
// For JSON responses from external APIs
const data = (await response.json()) as any;
// For createOpenAIFunctionsAgent with complex tool types
// @ts-ignore - Zod tuple schemas cause deep type recursion
const agent = await createOpenAIFunctionsAgent({ llm, tools, prompt });
This doesn't affect runtime validation (Zod still validates at runtime) - it only helps TypeScript's type checker avoid infinite recursion during compilation.
Use case: Building domain-specific AI applications (Zillow-style, TripAdvisor-style)
interface MCPTool {
name: string;
description: string;
inputSchema: any;
}
class CustomMapboxAgent {
private url = 'https://mcp.mapbox.com/mcp';
private headers: Record<string, string>;
private tools: Map<string, MCPTool> = new Map();
constructor(token?: string) {
const mapboxToken = token || process.env.MAPBOX_ACCESS_TOKEN;
this.headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${mapboxToken}`
};
}
async initialize() {
// Discover available tools from MCP server
await this.discoverTools();
}
private async discoverTools() {
const request = {
jsonrpc: '2.0',
id: 1,
method: 'tools/list'
};
const response = await this.sendMCPRequest(request);
response.result.tools.forEach((tool: MCPTool) => {
this.tools.set(tool.name, tool);
});
}
async callTool(toolName: string, params: any): Promise<any> {
const request = {
jsonrpc: '2.0',
id: Date.now(),
method: 'tools/call',
params: { name: toolName, arguments: params }
};
const response = await this.sendMCPRequest(request);
return response.result.content[0].text;
}
private async sendMCPRequest(request: any): Promise<any> {
const response = await fetch(this.url, {
method: 'POST',
headers: this.headers,
body: JSON.stringify(request)
});
const data = await response.json();
if (data.error) {
throw new Error(data.error.message);
}
return data;
}
// Domain-specific methods
async findPropertiesWithCommute(
homeLocation: [number, number],
workLocation: [number, number],
maxCommuteMinutes: number
) {
// Get isochrone from work location
const isochrone = await this.callTool('isochrone_tool', {
coordinates: { longitude: workLocation[0], latitude: workLocation[1] },
contours_minutes: [maxCommuteMinutes],
profile: 'mapbox/driving-traffic'
});
// Check if home is within isochrone
const isInRange = await this.callTool('point_in_polygon_tool', {
point: { longitude: homeLocation[0], latitude: homeLocation[1] },
polygon: JSON.parse(isochrone).features[0].geometry
});
return JSON.parse(isInRange);
}
async findRestaurantsNearby(location: [number, number], radiusMiles: number) {
// Search restaurants
const results = await this.callTool('category_search_tool', {
category: 'restaurant',
proximity: { longitude: location[0], latitude: location[1] }
});
// Filter by distance
const restaurants = JSON.parse(results);
const filtered = [];
for (const restaurant of restaurants) {
const distance = await this.callTool('distance_tool', {
from: { longitude: location[0], latitude: location[1] },
to: { longitude: restaurant.coordinates[0], latitude: restaurant.coordinates[1] },
units: 'miles'
});
if (parseFloat(distance) <= radiusMiles) {
filtered.push({
...restaurant,
distance: parseFloat(distance)
});
}
}
return filtered.sort((a, b) => a.distance - b.distance);
}
}
// Usage in Zillow-style app
const agent = new CustomMapboxAgent();
await agent.initialize();
const properties = await agent.findPropertiesWithCommute(
[-122.4194, 37.7749], // Home in SF
[-122.4, 37.79], // Work downtown
30 // Max 30min commute
);
// Usage in TripAdvisor-style app
const restaurants = await agent.findRestaurantsNearby(
[-73.9857, 40.7484], // Times Square
0.5 // Within 0.5 miles
);
Benefits:
┌─────────────────────────────────────┐
│ Your Application │
│ (Next.js, Express, FastAPI, etc.) │
└────────────────┬────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ AI Agent Layer │
│ (pydantic-ai, mastra, custom) │
└────────────────┬────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ Mapbox MCP Server │
│ (Geospatial tools abstraction) │
└────────────────┬────────────────────┘
│
┌──────┴──────┐
▼ ▼
┌─────────┐ ┌──────────┐
│ Turf.js │ │ Mapbox │
│ (Local) │ │ APIs │
└─────────┘ └──────────┘
Benefits:
You can use MCP for AI agent features while using direct Mapbox APIs for other parts of your app.
class GeospatialService {
constructor(
private mcpServer: MapboxMCPServer, // For AI features
private mapboxSdk: MapboxSDK // For direct app features
) {}
// AI Agent Feature: Natural language search
async aiSearchNearby(userQuery: string): Promise<string> {
// Let AI agent use MCP tools to interpret query and find places
// Returns natural language response
return await this.agent.execute(userQuery, [
this.mcpServer.tools.category_search_tool,
this.mcpServer.tools.directions_tool
]);
}
// Direct App Feature: Display route on map
async getRouteGeometry(origin: Point, dest: Point): Promise<LineString> {
// Direct API call for map rendering - returns GeoJSON
const result = await this.mapboxSdk.directions.getDirections({
waypoints: [origin, dest],
geometries: 'geojson'
});
return result.routes[0].geometry;
}
// Offline Feature: Distance calculations (always use MCP/Turf.js)
async calculateDistance(from: Point, to: Point): Promise<number> {
// No API cost, instant
return await this.mcpServer.callTool('distance_tool', {
from,
to,
units: 'miles'
});
}
}
Architecture Decision Guide:
| Use Case | Use This | Why |
|---|---|---|
| AI agent natural language features | MCP Server | Simplified tool interface, AI-friendly responses |
| Map rendering, direct UI controls | Mapbox SDK | More control, better performance |
| Distance/area calculations | MCP Server (offline tools) | Free, instant, no API calls |
| Custom map styling | Mapbox SDK | Fine-grained style control |
| Conversational geospatial queries | MCP Server | AI agent can chain tools |
// Find properties with good commute
async findPropertiesByCommute(
searchArea: Polygon,
workLocation: Point,
maxCommuteMinutes: number
) {
// 1. Get isochrone from work
const reachableArea = await mcp.callTool('isochrone_tool', {
coordinates: { longitude: workLocation[0], latitude: workLocation[1] },
contours_minutes: [maxCommuteMinutes],
profile: 'mapbox/driving'
});
// 2. Check each property
const propertiesInRange = [];
for (const property of properties) {
const inRange = await mcp.callTool('point_in_polygon_tool', {
point: { longitude: property.location[0], latitude: property.location[1] },
polygon: reachableArea
});
if (inRange) {
// 3. Get exact commute time
const directions = await mcp.callTool('directions_tool', {
coordinates: [property.location, workLocation],
routing_profile: 'mapbox/driving-traffic'
});
propertiesInRange.push({
...property,
commuteTime: directions.duration / 60
});
}
}
return propertiesInRange;
}
// Check if restaurant can deliver to address
async canDeliver(
restaurantLocation: Point,
deliveryAddress: Point,
maxDeliveryTime: number
) {
// 1. Calculate delivery zone
const deliveryZone = await mcp.callTool('isochrone_tool', {
coordinates: restaurantLocation,
contours_minutes: [maxDeliveryTime],
profile: 'mapbox/driving'
});
// 2. Check if address is in zone
const canDeliver = await mcp.callTool('point_in_polygon_tool', {
point: deliveryAddress,
polygon: deliveryZone
});
if (!canDeliver) return false;
// 3. Get accurate delivery time
const route = await mcp.callTool('directions_tool', {
coordinates: [restaurantLocation, deliveryAddress],
routing_profile: 'mapbox/driving-traffic'
});
return {
canDeliver: true,
estimatedTime: route.duration / 60,
distance: route.distance
};
}
// Build day itinerary with travel times
async buildItinerary(
hotel: Point,
attractions: Array<{name: string, location: Point}>
) {
// 1. Calculate distances from hotel
const attractionsWithDistance = await Promise.all(
attractions.map(async (attr) => ({
...attr,
distance: await mcp.callTool('distance_tool', {
from: hotel,
to: attr.location,
units: 'miles'
})
}))
);
// 2. Get travel time matrix
const matrix = await mcp.callTool('matrix_tool', {
origins: [hotel],
destinations: attractions.map(a => a.location),
profile: 'mapbox/walking'
});
// 3. Sort by walking time
return attractionsWithDistance
.map((attr, idx) => ({
...attr,
walkingTime: matrix.durations[0][idx] / 60
}))
.sort((a, b) => a.walkingTime - b.walkingTime);
}
class CachedMapboxMCP {
private cache = new Map<string, { result: any; timestamp: number }>();
private cacheTTL = 3600000; // 1 hour
async callTool(name: string, params: any): Promise<any> {
// Cache offline tools indefinitely (deterministic)
const offlineTools = ['distance_tool', 'point_in_polygon_tool', 'bearing_tool'];
const ttl = offlineTools.includes(name) ? Infinity : this.cacheTTL;
// Check cache
const cacheKey = JSON.stringify({ name, params });
const cached = this.cache.get(cacheKey);
if (cached && Date.now() - cached.timestamp < ttl) {
return cached.result;
}
// Call MCP
const result = await this.mcpServer.callTool(name, params);
// Store in cache
this.cache.set(cacheKey, {
result,
timestamp: Date.now()
});
return result;
}
}
// ❌ Bad: Sequential calls
for (const location of locations) {
const distance = await mcp.callTool('distance_tool', {
from: userLocation,
to: location
});
}
// ✅ Good: Parallel batch
const distances = await Promise.all(
locations.map((location) =>
mcp.callTool('distance_tool', {
from: userLocation,
to: location
})
)
);
// ✅ Better: Use matrix tool
const matrix = await mcp.callTool('matrix_tool', {
origins: [userLocation],
destinations: locations
});
Clear, specific tool descriptions are critical for helping LLMs select the right tools. Poor descriptions lead to incorrect tool calls, wasted API requests, and user frustration.
Problem: "How far is it from A to B?" - Could trigger either directions_tool OR distance_tool
// ❌ Ambiguous descriptions
{
name: 'directions_tool',
description: 'Get directions between two locations' // Could mean distance
}
{
name: 'distance_tool',
description: 'Calculate distance between two points' // Unclear what kind
}
// ✅ Clear, specific descriptions
{
name: 'directions_tool',
description: 'Get turn-by-turn driving directions with traffic-aware route distance and travel time. Use when you need the actual route, navigation instructions, or driving duration. Returns route geometry, distance along roads, and time estimate.'
}
{
name: 'distance_tool',
description: 'Calculate straight-line (great-circle) distance between two points. Use for quick "as the crow flies" distance checks, proximity comparisons, or when routing is not needed. Works offline, instant, no API cost.'
}
Problem: "Find coffee shops nearby" - Could trigger category_search_tool OR search_and_geocode_tool
// ❌ Ambiguous
{
name: 'search_poi',
description: 'Search for places'
}
// ✅ Clear when to use each
{
name: 'category_search_tool',
description: 'Find ALL places of a specific type/category (e.g., "all coffee shops", "restaurants", "gas stations") near a location. Use for browsing or discovering places by category. Returns multiple results.'
}
{
name: 'search_and_geocode_tool',
description: 'Search for a SPECIFIC named place or address (e.g., "Starbucks on Main St", "123 Market St"). Use when the user provides a business name, street address, or landmark. Returns best match.'
}
Problem: "Where can I go in 15 minutes?" - Could trigger isochrone_tool OR directions_tool
// ❌ Confusing
{
name: 'isochrone_tool',
description: 'Calculate travel time area'
}
// ✅ Clear distinction
{
name: 'isochrone_tool',
description: 'Calculate the AREA reachable within a time limit from a starting point. Returns a GeoJSON polygon showing everywhere you can reach. Use for: "What can I reach in X minutes?", service area analysis, catchment zones, delivery zones.'
}
{
name: 'directions_tool',
description: 'Get route from point A to specific point B. Returns turn-by-turn directions to ONE destination. Use for: "How do I get to X?", "Route from A to B", navigation to a known destination.'
}
// ✅ Complete example
const searchPOITool = new DynamicStructuredTool({
name: 'category_search_tool',
description: `Find places by category type (restaurants, hotels, coffee shops, gas stations, etc.) near a location.
Use when the user wants to:
- Browse places of a certain type: "coffee shops nearby", "find restaurants"
- Discover options: "what hotels are in this area?"
- Search by industry/amenity, not by specific name
Returns: List of matching places with names, addresses, and coordinates.
DO NOT use for:
- Specific named places (use search_and_geocode_tool instead)
- Addresses (use search_and_geocode_tool or reverse_geocode_tool)`
// ... schema and implementation
});
Add tool selection guidance to your agent's system prompt:
const systemPrompt = `You are a location intelligence assistant.
TOOL SELECTION RULES:
- Use distance_tool for straight-line distance ("as the crow flies")
- Use directions_tool for route distance along roads with traffic
- Use category_search_tool for finding types of places ("coffee shops")
- Use search_and_geocode_tool for specific addresses or named places ("123 Main St", "Starbucks downtown")
- Use isochrone_tool for "what can I reach in X minutes" questions
- Use offline tools (distance_tool, point_in_polygon_tool) when real-time data is not needed
When in doubt, prefer:
1. Offline tools over API calls (faster, free)
2. Specific tools over general ones
3. Asking for clarification over guessing`;
// Use offline tools when possible (faster, free)
const localOps = {
distance: 'distance_tool', // Turf.js
pointInPolygon: 'point_in_polygon_tool', // Turf.js
bearing: 'bearing_tool', // Turf.js
area: 'area_tool' // Turf.js
};
// Use API tools when necessary (requires token, slower)
const apiOps = {
directions: 'directions_tool', // Mapbox API
geocoding: 'reverse_geocode_tool', // Mapbox API
isochrone: 'isochrone_tool', // Mapbox API
search: 'category_search_tool' // Mapbox API
};
// Choose based on requirements
function chooseTool(operation: string, needsRealtime: boolean) {
if (needsRealtime) {
return apiOps[operation]; // Traffic, live data
}
return localOps[operation] || apiOps[operation];
}
class RobustMapboxMCP {
async callToolWithRetry(name: string, params: any, maxRetries: number = 3): Promise<any> {
for (let i = 0; i < maxRetries; i++) {
try {
return await this.mcpServer.callTool(name, params);
} catch (error) {
if (error.code === 'RATE_LIMIT') {
// Exponential backoff
await this.sleep(Math.pow(2, i) * 1000);
continue;
}
if (error.code === 'INVALID_TOKEN') {
// Non-retryable error
throw error;
}
if (i === maxRetries - 1) {
throw error;
}
}
}
}
async callToolWithFallback(primaryTool: string, fallbackTool: string, params: any): Promise<any> {
try {
return await this.callTool(primaryTool, params);
} catch (error) {
console.warn(`Primary tool ${primaryTool} failed, using fallback`);
return await this.callTool(fallbackTool, params);
}
}
}
// ✅ Good: Use environment variables
const mcp = new MapboxMCP({
token: process.env.MAPBOX_ACCESS_TOKEN
});
// ❌ Bad: Hardcode tokens
const mcp = new MapboxMCP({
token: 'pk.ey...' // Never do this!
});
// ✅ Good: Use scoped tokens
// Create token with minimal scopes:
// - directions:read
// - geocoding:read
// - No write permissions
class RateLimitedMCP {
private requestQueue: Array<() => Promise<any>> = [];
private requestsPerMinute = 300;
private currentMinute = Math.floor(Date.now() / 60000);
private requestCount = 0;
async callTool(name: string, params: any): Promise<any> {
// Check rate limit
const minute = Math.floor(Date.now() / 60000);
if (minute !== this.currentMinute) {
this.currentMinute = minute;
this.requestCount = 0;
}
if (this.requestCount >= this.requestsPerMinute) {
// Wait until next minute
const waitMs = (this.currentMinute + 1) * 60000 - Date.now();
await this.sleep(waitMs);
}
this.requestCount++;
return await this.mcpServer.callTool(name, params);
}
}
// Mock MCP server for testing
class MockMapboxMCP {
async callTool(name: string, params: any): Promise<any> {
const mocks = {
distance_tool: () => '2.5',
directions_tool: () => JSON.stringify({
duration: 1200,
distance: 5000,
geometry: {...}
}),
point_in_polygon_tool: () => 'true'
};
return mocks[name]?.() || '{}';
}
}
// Use in tests
describe('Property search', () => {
it('finds properties within commute time', async () => {
const agent = new CustomMapboxAgent(new MockMapboxMCP());
const results = await agent.findPropertiesWithCommute(
[-122.4, 37.7],
[-122.41, 37.78],
30
);
expect(results).toHaveLength(5);
});
});
Invoke this skill when:
Weekly Installs
120
Repository
GitHub Stars
35
First Seen
Mar 5, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykWarn
Installed on
github-copilot115
opencode114
gemini-cli113
cline113
codex113
cursor113
超能力技能使用指南:AI助手技能调用优先级与工作流程详解
52,100 周安装
Intercom自动化指南:通过Rube MCP与Composio实现客户支持对话管理
69 周安装
二进制初步分析指南:使用ReVa工具快速识别恶意软件与逆向工程
69 周安装
PrivateInvestigator 道德人员查找工具 | 公开数据调查、反向搜索与背景研究
69 周安装
TorchTitan:PyTorch原生分布式大语言模型预训练平台,支持4D并行与H100 GPU加速
69 周安装
screenshot 截图技能:跨平台桌面截图工具,支持macOS/Linux权限管理与多模式捕获
69 周安装
tmux进程管理最佳实践:交互式Shell初始化、会话命名与生命周期管理
69 周安装