from typing import Any
import httpx, os
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("search")
async def make_request(url: str) -> dict[str, Any] | None:
headers = {}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
#response.raise_for_status()
return response.json()
except Exception:
return None
@mcp.tool()
async def get_ip() -> str | str:
"""
使用第三方服务获取公网IP地址。
Returns:
str | str: 返回公网IP地址的字符串，如果请求失败则返回错误信息字符串。
"""
url = 'https://ipinfo.io/json'
data = await make_request(url)
if not data:
return "Failed to obtain public IP address."
else:
return data["ip"]
@mcp.tool()
async def get_location_ip(ip: str) -> str | str:
"""
使用百度地图API根据公网IP地址获取地理位置信息。
Args:
ip (str): 公网IP地址。
Returns:
str | str: 返回包含地理位置信息的字符串，如果请求失败则返回错误信息字符串。
"""
url = "https://api.map.baidu.com/location/ip?"
ak = os.getenv("BAIDU_MAP_API_KEY")
param = {
"ak": ak,
"ip": ip,
"coor": "bd09ll"
}
full_url = httpx.URL(url, params=param)
data = await make_request(str(full_url))
if not data or data.get("status") != 0:
return "Failed to obtain location based on public IP address."
else:
return f"address: {data['content']['address']}, lng: {data['content']['point']['x']}, lat: {data['content']['point']['y']}"
@mcp.tool()
async def get_location_text(address: str) -> str | str:
"""
使用百度地图API根据地址获取地理位置信息。
Args:
address (str): 具体地址，例如 "北京市海淀区中关村大街27号"。
Returns:
str | str: 返回包含地理位置信息的字符串，如果请求失败则返回错误信息字符串。
"""
url = "https://api.map.baidu.com/geocoding/v3/?"
ak = os.getenv("BAIDU_MAP_API_KEY")
param = {
"address": address,
"output": "json",
"ak": ak
}
full_url = httpx.URL(url, params=param)
data = await make_request(str(full_url))
if not data or data.get("status") != 0:
return "Failed to obtain geographic location information."
else:
return f"lng: {data['result']['location']['lng']}, lat: {data['result']['location']['lat']}"
@mcp.tool()
async def get_places(query: str, location: str, radius: int = 1000) -> str | str:
"""
使用百度地图API搜索指定位置附近的地点。
Args:
query (str): 搜索关键词，例如 "天安门"。
radius (int): 搜索半径，单位为米，默认值为1000米。
location (str): 中心点的经纬度，格式为 "纬度,经度"。
Returns:
str | str: 返回搜索结果的字符串，如果请求失败则返回错误信息字符串。
"""
url = "https://api.map.baidu.com/place/v2/search?"
ak = os.getenv("BAIDU_MAP_API_KEY")
param = {
"query": query,
"location": location,
"radius": radius,
"output": "json",
"ak": ak,
"scope": 1
}
full_url = httpx.URL(url, params=param)
data = await make_request(str(full_url))
if not data or data.get("status") != 0:
return "Location search failed."
elif data['results'] == []:
return "No results found.(Nearby is relatively remote or not in the Chinese region)"
else:
return str(data['results'])
@mcp.tool()
async def get_place_driving_distance(origin: str, destination: str) -> str | str:
"""
使用百度地图API获取地点之间的距离和需要的时间，方式为驾驶。
Args:
origin (str): 起点的经纬度，格式为 "纬度,经度"。
destination (str): 终点的经纬度，格式为 "纬度,经度"。
Returns:
str | str: 返回地点之间的距离和需要的时间的字符串，如果请求失败则返回错误信息字符串。
"""
url = "https://api.map.baidu.com/directionlite/v1/driving?"
ak = os.getenv("BAIDU_MAP_API_KEY")
param = {
"origin": origin,
"destination": destination,
"ak": ak
}
full_url = httpx.URL(url, params=param)
data = await make_request(str(full_url))
if not data or data.get("status") != 0:
return "Failed to obtain distance and time."
elif data['status'] == 2:
return "Failed to obtain distance and time, maybe not in the Chinese region."
else:
return f"distance: {data['result']['routes'][0]['distance']} meters, duration: {data['result']['routes'][0]['duration']} seconds, steps: {str(data['result']['routes'][0]['steps'])}"
@mcp.tool()
async def get_place_riding_distance(origin: str, destination: str) -> str | str:
"""
使用百度地图API获取地点之间的距离和需要的时间，方式为骑行。
Args:
origin (str): 起点的经纬度，格式为 "纬度,经度"。
destination (str): 终点的经纬度，格式为 "纬度,经度"。
Returns:
str | str: 返回地点之间的距离和需要的时间的字符串，如果请求失败则返回错误信息字符串。
"""
url = "https://api.map.baidu.com/directionlite/v1/riding?"
ak = os.getenv("BAIDU_MAP_API_KEY")
param = {
"origin": origin,
"destination": destination,
"ak": ak
}
full_url = httpx.URL(url, params=param)
data = await make_request(str(full_url))
if not data or data.get("status") != 0:
return "Failed to obtain distance and time."
elif data['status'] == 2:
return "Failed to obtain distance and time, maybe not in the Chinese region."
else:
return f"distance: {data['result']['routes'][0]['distance']} meters, duration: {data['result']['routes'][0]['duration']} seconds, steps: {str(data['result']['routes'][0]['steps'])}"
@mcp.tool()
async def get_place_walking_distance(origin: str, destination: str) -> str | str:
"""
使用百度地图API获取地点之间的距离和需要的时间，方式为步行。
Args:
origin (str): 起点的经纬度，格式为 "纬度,经度"。
destination (str): 终点的经纬度，格式为 "纬度,经度"。
Returns:
str | str: 返回地点之间的距离和需要的时间的字符串，如果请求失败则返回错误信息字符串。
"""
url = "https://api.map.baidu.com/directionlite/v1/walking?"
ak = os.getenv("BAIDU_MAP_API_KEY")
param = {
"origin": origin,
"destination": destination,
"ak": ak
}
full_url = httpx.URL(url, params=param)
data = await make_request(str(full_url))
if not data or data.get("status") != 0:
return "Failed to obtain distance and time."
elif data['status'] == 2:
return "Failed to obtain distance and time, maybe not in the Chinese region."
else:
return f"distance: {data['result']['routes'][0]['distance']} meters, duration: {data['result']['routes'][0]['duration']} seconds, steps: {str(data['result']['routes'][0]['steps'])}"
@mcp.tool()
async def get_place_transit_distance(origin: str, destination: str) -> str | str:
"""
使用百度地图API获取地点之间的距离和需要的时间，方式为公共交通。
Args:
origin (str): 起点的经纬度，格式为 "纬度,经度"。
destination (str): 终点的经纬度，格式为 "纬度,经度"。
Returns:
str | str: 返回地点之间的距离和需要的时间的字符串，如果请求失败则返回错误信息字符串。
"""
url = "https://api.map.baidu.com/directionlite/v1/transit?"
ak = os.getenv("BAIDU_MAP_API_KEY")
param = {
"origin": origin,
"destination": destination,
"ak": ak
}
full_url = httpx.URL(url, params=param)
data = await make_request(str(full_url))
if not data or data.get("status") != 0:
return "Failed to obtain distance and time."
elif data['status'] == 2:
return "Failed to obtain distance and time, maybe not in the Chinese region."
else:
return f"distance: {data['result']['routes'][0]['distance']} meters, duration: {data['result']['routes'][0]['duration']} seconds, steps: {str(data['result']['routes'][0]['steps'])}"
if __name__ == "__main__":
mcp.run(transport='stdio')
