https://mcp.mapbox.com/mcp

Authorization: Bearer your_mapbox_token

npm install @mapbox/mcp-server

npx @mapbox/mcp-server

export MAPBOX_ACCESS_TOKEN="your_token_here"

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']

undefined

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

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']

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 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 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

Use search_poi for finding types of places (restaurants, hotels).
Use reverse_geocode_tool for converting coordinates to addresses.""",
tools=[GeocodeTool(), SearchPOITool()],
verbose=True

Use directions_tool for route distance along roads with traffic.
Always use when traffic-aware travel time is needed.""",
tools=[DirectionsTool()],
verbose=True

**Real-world example - Restaurant finder crew:**

```python

    # 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()

**Benefits:**

- Multi-agent orchestration with geospatial tools
- Task dependencies and context passing
- Role-based agent specialization
- Autonomous crew execution

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']

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'
    })

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
    })

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]}
    })

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
    })

**Real-world example - Property search agent:**

```python
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)

**Benefits:**

- Lightweight and efficient
- Simple tool definition
- Code-based agent execution
- Great for production deployment

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)' }
]);

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'
});

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 });

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
);