ZhouYI/05_MCP/travel_agent/tools.py

from langchain_mcp_adapters.client import MultiServerMCPClient

import asyncio
mcp_client=MultiServerMCPClient(
    {
        "amap": {
      "transport": "streamable_http",
      "url": "https://mcp.api-inference.modelscope.net/75ce1803fffa45/mcp"
    },
    
    
 
    "hotel-recommend": {
      "transport": "streamable_http",
      "url": "https://mcp.api-inference.modelscope.net/0c1ea913c4ef4f/mcp"
    }
}
)
_mcp_all_tools=None
_tools_lock = asyncio.Lock()

async def get_tools():
    global _mcp_all_tools
    if _mcp_all_tools is None:
        async with _tools_lock:
            if _mcp_all_tools is None:
                _mcp_all_tools = await mcp_client.get_tools()
    return _mcp_all_tools


async def get_transport_tools():
    """获取地图交通工具"""
    all_mcp_tools=await get_tools()
    transport_tools_names=["maps_geo","maps_direction_transit_integrated","maps_direction_driving","maps_direction_walking","maps_bicycling","maps_distance"]
    transport_tools=[tool for tool in all_mcp_tools if tool.name in transport_tools_names]
    return transport_tools

async def get_hotel_tools():
    """获取酒店工具"""
    all_mcp_tools=await get_tools()
    hotel_tools_names=["hotel_search_and_recommend"]
    hotel_tools=[tool for tool in all_mcp_tools if tool.name in hotel_tools_names]
    return hotel_tools

async def get_attraction_tools():
    """获取POI信息工具"""
    all_mcp_tools=await get_tools()
    attraction_tools_names=["maps_geo","maps_text_search","maps_around_search","maps_search_detail","maps_weather"]
    attraction_tools=[tool for tool in all_mcp_tools if tool.name in attraction_tools_names]
    return attraction_tools

async def main():
    tools = await get_tools()
    print(f"可用工具：{[tool.name for tool in tools]}\n")



if __name__ == "__main__":
    asyncio.run(main())