#!/usr/bin/env python3
"""
Japanese Weather MCP Server
MCPサーバーとして動作し、日本の天気情報を提供する
"""
import asyncio
import logging
import sys
from typing import Any, Dict, List, Optional
from fastmcp import Context, FastMCP
from pydantic import BaseModel
from .config import get_settings
from .location_resolver import LocationResolver
from .weather_service import WeatherService
settings = get_settings()
# ログ設定
logging.basicConfig(
level=getattr(logging, settings.log_level), format=settings.log_format
)
logger = logging.getLogger(__name__)
class WeatherRequest(BaseModel):
location: str
days: Optional[int] = 1
class WeatherResponse(BaseModel):
location: str
weather: str
temperature: float
humidity: int
precipitation_probability: int
date: str
mcp = FastMCP("Japanese Weather")
weather_service = WeatherService()
location_resolver = LocationResolver()
@mcp.tool()
async def get_weather(
context: Context, location: str, days: int = 1
) -> List[Dict[str, Any]]:
"""
日本の指定された地域の天気情報を取得する
Args:
location: 地域名（都道府県名、市区町村名）
days: 取得する日数（1-7日、デフォルト1日）
Returns:
天気情報のリスト
"""
logger.info(f"天気情報を取得: location={location}, days={days}")
try:
# 地名を緯度経度に変換
lat, lon = await location_resolver.resolve_location(location)
logger.debug(f"地名解決結果: {location} -> ({lat}, {lon})")
# 天気情報を取得
weather_data = await weather_service.get_weather(lat, lon, days)
logger.info(f"天気情報取得成功: {len(weather_data)}件")
return weather_data
except Exception as e:
error_msg = f"天気情報の取得に失敗しました: {str(e)}"
logger.error(error_msg)
return [{"error": error_msg}]
@mcp.tool()
async def search_locations(context: Context, query: str) -> List[Dict[str, str]]:
"""
地域名を検索する
Args:
query: 検索クエリ（地域名の一部）
Returns:
マッチした地域のリスト
"""
logger.info(f"地域検索: query={query}")
try:
locations = await location_resolver.search_locations(query)
logger.info(f"地域検索結果: {len(locations)}件")
return locations
except Exception as e:
error_msg = f"地域検索に失敗しました: {str(e)}"
logger.error(error_msg)
return [{"error": error_msg}]
@mcp.tool()
async def get_current_weather(context: Context, location: str) -> Dict[str, Any]:
"""
日本の指定された地域の現在の天気情報を取得する
Args:
location: 地域名（都道府県名、市区町村名）
Returns:
現在の天気情報
"""
logger.info(f"現在の天気情報を取得: location={location}")
try:
# 地名を緯度経度に変換
lat, lon = await location_resolver.resolve_location(location)
logger.debug(f"地名解決結果: {location} -> ({lat}, {lon})")
# 現在の天気情報を取得
current_weather = await weather_service.get_current_weather(lat, lon)
logger.info("現在の天気情報取得成功")
return current_weather
except Exception as e:
error_msg = f"現在の天気情報の取得に失敗しました: {str(e)}"
logger.error(error_msg)
return {"error": error_msg}
def main():
"""MCPサーバーを起動"""
import os
# 環境変数からトランスポート方式を決定
transport = os.getenv("MCP_TRANSPORT", "streamable-http")
host = os.getenv("MCP_HOST", "127.0.0.1")
port = int(os.getenv("MCP_PORT", "8000"))
logger.info(
f"Japanese Weather MCP Server を起動しています... (transport: {transport})"
)
try:
if transport == "streamable-http":
path = os.getenv("MCP_PATH", "/mcp/")
logger.info(f"HTTPサーバーを起動: http://{host}:{port}{path}")
mcp.run(
transport="streamable-http",
host=host,
port=port,
path=path,
log_level="debug",
)
elif transport == "sse":
path = os.getenv("MCP_PATH", "/mcp/")
logger.info(f"SSEサーバーを起動: http://{host}:{port}{path}")
mcp.run(transport="sse", host=host, port=port, path=path, log_level="debug")
else:
logger.info("STDIOトランスポートで起動")
mcp.run(transport="stdio")
except Exception as e:
logger.error(f"サーバー起動に失敗: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
main()
