OCI生成AIサービスで指定した地域のラーメン屋情報を教えてくれるMCPサーバを作ってみた #Python – Qiita

import os
from dotenv import load_dotenv
from typing import Annotated
import httpx
from mcp.server.fastmcp import FastMCP

# ホットペッパーグルメwebサービス利用に必要なAPIキーを.envファイルから取得
load_dotenv()
API_KEY = os.environ['API_KEY']

# FastMCPサーバを初期化
mcp = FastMCP("ramen")

# APIエンドポイント、エージェント名を設定
API_BASE = "http://webservice.recruit.co.jp/hotpepper"
USER_AGENT = "ramen-app/1.0"

# MCPサーバを定義
@mcp.tool()
async def get_small_area_code(
    area_name: Annotated[str, "小エリアコードを取得したい市区名 (例: 京都市)"]
) -> str:
    """
    リクルートWebサービスから市区名をもとに小エリアコードを取得する。
    """
    url = f"{API_BASE}/small_area/v1/?key={API_KEY}&format=json"
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            area_code = ""

            for each_area in response.json()["results"]["small_area"]:
                if area_name in each_area["name"]:
                    area_code = each_area["code"]
                    break
            return area_code

        except Exception as e:
            return f"データ取得中にエラーが発生しました: {repr(e)}"

@mcp.tool()
async def get_gourmet_info(
        small_area_code: Annotated[str, "ラーメン屋の情報を取得したい小エリアのコード (例: X010)"]
) -> str:
    """
    リクルートWebサービスから小エリアコードをもとに指定地域のラーメン屋情報を取得する。
    """
    url = f"{API_BASE}/gourmet/v1/?key={API_KEY}&small_area={small_area_code}&genre=G013&format=json"
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            shop_info_list = ""

            for each_shop in response.json()["results"]["shop"]:
                shop_info_list = shop_info_list + f"""
店名: {each_shop["name"]}　{each_shop["name_kana"]}
特徴: {each_shop["catch"]}

"""
            return shop_info_list

        except Exception as e:
            return f"データ取得中にエラーが発生しました: {repr(e)}"
    
if __name__ == "__main__":
    mcp.run(transport="stdio")