Add ruff and apply it
This commit is contained in:
parent
1604073254
commit
2af8046eb7
1 changed files with 87 additions and 14 deletions
|
|
@ -1,4 +1,5 @@
|
|||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
|
|
@ -9,12 +10,10 @@ mcp = FastMCP("weather")
|
|||
NWS_API_BASE = "https://api.weather.gov"
|
||||
USER_AGENT = "weather-app/1.0"
|
||||
|
||||
|
||||
async def make_nws_request(url: str) -> dict[str, Any] | None:
|
||||
"""Make a request to the NWS API with proper error handling."""
|
||||
headers = {
|
||||
"User-Agent": USER_AGENT,
|
||||
"Accept": "application/geo+json"
|
||||
}
|
||||
headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
|
||||
async with httpx.AsyncClient() as client:
|
||||
try:
|
||||
response = await client.get(url, headers=headers, timeout=30.0)
|
||||
|
|
@ -23,17 +22,19 @@ async def make_nws_request(url: str) -> dict[str, Any] | None:
|
|||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def format_alert(feature: dict) -> str:
|
||||
"""Format an alert feature into a readable string."""
|
||||
props = feature["properties"]
|
||||
return f"""
|
||||
Event: {props.get('event', 'Unknown')}
|
||||
Area: {props.get('areaDesc', 'Unknown')}
|
||||
Severity: {props.get('severity', 'Unknown')}
|
||||
Description: {props.get('description', 'No description available')}
|
||||
Instructions: {props.get('instruction', 'No specific instructions provided')}
|
||||
Event: {props.get("event", "Unknown")}
|
||||
Area: {props.get("areaDesc", "Unknown")}
|
||||
Severity: {props.get("severity", "Unknown")}
|
||||
Description: {props.get("description", "No description available")}
|
||||
Instructions: {props.get("instruction", "No specific instructions provided")}
|
||||
"""
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def get_alerts(state: str) -> str:
|
||||
"""Get weather alerts for a US state.
|
||||
|
|
@ -53,6 +54,7 @@ async def get_alerts(state: str) -> str:
|
|||
alerts = [format_alert(feature) for feature in data["features"]]
|
||||
return "\n---\n".join(alerts)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def get_forecast(latitude: float, longitude: float) -> str:
|
||||
"""Get weather forecast for a location.
|
||||
|
|
@ -80,18 +82,20 @@ async def get_forecast(latitude: float, longitude: float) -> str:
|
|||
forecasts = []
|
||||
for period in periods[:5]: # Only show next 5 periods
|
||||
forecast = f"""
|
||||
{period['name']}:
|
||||
Temperature: {period['temperature']}°{period['temperatureUnit']}
|
||||
Wind: {period['windSpeed']} {period['windDirection']}
|
||||
Forecast: {period['detailedForecast']}
|
||||
{period["name"]}:
|
||||
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
|
||||
Wind: {period["windSpeed"]} {period["windDirection"]}
|
||||
Forecast: {period["detailedForecast"]}
|
||||
"""
|
||||
forecasts.append(forecast)
|
||||
|
||||
return "\n---\n".join(forecasts)
|
||||
|
||||
|
||||
def main():
|
||||
# Initialize and run the server
|
||||
mcp.run(transport='stdio')
|
||||
mcp.run(transport="stdio")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
Loading…
Reference in a new issue