main.py•4.58 kB
# app/main.py
from fastapi import FastAPI
# For more specific HTTP error responses outside MCP format, you might use:
# from fastapi import FastAPI, HTTPException
# Import our models, service function, and custom error
from .models import MCPRequest, MCPResponse, MCPWeatherData, MCPRequestParameters
from .services import fetch_weather_data_from_provider, WeatherServiceError
# The settings are used within services.py, so not directly needed here unless for other purposes.
# from .core.config import get_settings
# Initialize FastAPI app
app = FastAPI(
title="MCP Weather Server",
description="A Model Context Protocol server for fetching weather information.",
version="1.0.0",
# You can add more metadata for OpenAPI docs here
# docs_url="/docs", # Default
# redoc_url="/redoc" # Default
)
@app.post(
"/mcp/weather",
response_model=MCPResponse, # Defines the expected response structure for documentation and validation
summary="Get Current Weather via MCP",
tags=["MCP Tools"] # Groups endpoints in the API docs
)
async def get_weather_mcp_endpoint(request: MCPRequest):
"""
MCP endpoint to get current weather information for a specified location.
Receives an MCPRequest, processes it using the weather_tool,
and returns an MCPResponse.
"""
# --- Validate MCP Request Specifics for this Endpoint ---
# Although MCPRequest has tool_id and method, this specific endpoint is only for weather.
# You might have a more generic /mcp endpoint in a larger system that routes based on tool_id.
if request.tool_id != "weather_tool":
return MCPResponse(
protocol_version=request.protocol_version, # Echo back protocol version
tool_id=request.tool_id,
status="error",
error_message=f"Invalid tool_id '{request.tool_id}'. This endpoint is dedicated to 'weather_tool'."
)
if request.method != "get_current_weather":
return MCPResponse(
protocol_version=request.protocol_version,
tool_id=request.tool_id,
status="error",
error_message=f"Invalid method '{request.method}'. This endpoint only supports 'get_current_weather' for the 'weather_tool'."
)
# --- Extract Parameters ---
# The request.parameters should already be validated by Pydantic as MCPRequestParameters
location = request.parameters.location
# --- Call Service Layer ---
try:
# The service function returns a dictionary of processed weather data
processed_weather_data_dict = await fetch_weather_data_from_provider(location)
# Create an MCPWeatherData Pydantic model instance from the dictionary
# This also validates that the service returned all required fields for MCPWeatherData
weather_data_for_mcp = MCPWeatherData(**processed_weather_data_dict)
# --- Construct Successful MCP Response ---
return MCPResponse(
protocol_version=request.protocol_version,
tool_id=request.tool_id,
status="success",
data=weather_data_for_mcp
)
except WeatherServiceError as e:
# Custom errors from our service layer
# In a real app, you'd log these errors server-side:
# logger.error(f"WeatherServiceError for location '{location}': {e} (Status Code: {e.status_code})")
return MCPResponse(
protocol_version=request.protocol_version,
tool_id=request.tool_id,
status="error",
error_message=str(e) # The message from WeatherServiceError
)
except Exception as e:
# Catch any other unexpected errors during the endpoint logic
# In a real app, you'd log these critical errors server-side:
# logger.critical(f"Unexpected error in get_weather_mcp_endpoint for location '{location}': {e}", exc_info=True)
# Return a generic error in MCP format
return MCPResponse(
protocol_version=request.protocol_version,
tool_id=request.tool_id, # Or a generic tool_id if request.tool_id is problematic
status="error",
error_message="An unexpected internal server error occurred while processing your request."
)
# Optional: A root endpoint for a simple health check or landing page for the API
@app.get("/", include_in_schema=False) # Hides from OpenAPI docs if desired
async def read_root():
return {"message": "Welcome to the MCP Weather Server! See /docs for API details."}