Building Intelligent Weather Agents with AWS Bedrock Agent Core and Model Context Protocol

import boto3
import json
from datetime import datetime, timedelta

class CognitoAuthSetup:
"""Configure Amazon Cognito for Agent Core authentication"""

def __init__(self, region: str = "us-east-1"):
self.cognito = boto3.client('cognito-idp', region_name=region)
self.iam = boto3.client('iam', region_name=region)

def create_user_pool(self, pool_name: str) -> str:
"""Create Cognito user pool with proper configuration"""

response = self.cognito.create_user_pool(
PoolName=pool_name,
Policies={
'PasswordPolicy': {
'MinimumLength': 12,
'RequireUppercase': True,
'RequireLowercase': True,
'RequireNumbers': True,
'RequireSymbols': False
}
},
Schema=[
{
'Name': 'email',
'AttributeDataType': 'String',
'Required': True,
'Mutable': False
},
{
'Name': 'agent_role',
'AttributeDataType': 'String',
'Mutable': True
}
],
AutoVerifiedAttributes=['email'],
UserAttributeUpdateSettings={
'AttributesRequireVerificationBeforUpdate': ['email']
}
)

user_pool_id = response['UserPool']['Id']
print(f"✓ Created Cognito User Pool: {user_pool_id}")

return user_pool_id

def create_app_client(self, user_pool_id: str,
client_name: str = "weather-agent") -> dict:
"""Create Cognito app client for Agent Core"""

response = self.cognito.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=client_name,
ExplicitAuthFlows=[
'ADMIN_USER_PASSWORD_AUTH',
'ALLOW_REFRESH_TOKEN_AUTH'
],
GenerateSecret=False, # Agent Core works better without secret
AllowedOAuthFlows=['client_credentials'],
AllowedOAuthScopes=['weather-agent/invoke'],
AccessTokenValidity=1,
IdTokenValidity=1,
RefreshTokenValidity=30,
TokenValidityUnits={
'AccessToken': 'hours',
'IdToken': 'hours',
'RefreshToken': 'days'
}
)

return {
'client_id': response['UserPoolClient']['ClientId'],
'user_pool_id': user_pool_id
}

def create_agent_user(self, user_pool_id: str,
email: str,
temporary_password: str = None) -> dict:
"""Create Cognito user for Agent Core service account"""

if not temporary_password:
temporary_password = self._generate_secure_password()

response = self.cognito.admin_create_user(
UserPoolId=user_pool_id,
Username=email,
UserAttributes=[
{'Name': 'email', 'Value': email},
{'Name': 'email_verified', 'Value': 'true'},
{'Name': 'agent_role', 'Value': 'weather-agent'}
],
TemporaryPassword=temporary_password,
MessageAction='SUPPRESS' # Don't send email
)

# Set permanent password programmatically
self.cognito.admin_set_user_password(
UserPoolId=user_pool_id,
Username=email,
Password=temporary_password,
Permanent=True
)

return {
'username': email,
'password': temporary_password,
'user_pool_id': user_pool_id
}

def get_jwt_token(self, user_pool_id: str, client_id: str,
username: str, password: str) -> str:
"""Obtain JWT token from Cognito for Agent Core"""

response = self.cognito.admin_initiate_auth(
UserPoolId=user_pool_id,
ClientId=client_id,
AuthFlow='ADMIN_USER_PASSWORD_AUTH',
AuthParameters={
'USERNAME': username,
'PASSWORD': password
}
)

id_token = response['AuthenticationResult']['IdToken']
access_token = response['AuthenticationResult']['AccessToken']

print(f"✓ Obtained JWT tokens (expires in 1 hour)")

return {
'id_token': id_token,
'access_token': access_token,
'token_type': 'Bearer'
}

@staticmethod
def _generate_secure_password() -> str:
"""Generate secure temporary password"""
import secrets
import string

chars = string.ascii_letters + string.digits + "!@#$%"
password = ''.join(secrets.choice(chars) for _ in range(16))

# Ensure it meets policy requirements
return f"Temp{secrets.randbelow(10000):04d}!@"

# Usage Example
if __name__ == "__main__":
setup = CognitoAuthSetup()

# Create user pool
user_pool_id = setup.create_user_pool("WeatherAgentPool")

# Create app client
app_client = setup.create_app_client(user_pool_id)

# Create agent user
agent_user = setup.create_agent_user(
user_pool_id,
"weather-agent@example.com"
)

# Get JWT token
tokens = setup.get_jwt_token(
user_pool_id,
app_client['client_id'],
agent_user['username'],
agent_user['password']
)

print(f"\nConfiguration Summary:")
print(f"User Pool ID: {user_pool_id}")
print(f"Client ID: {app_client['client_id']}")
print(f"Agent Username: {agent_user['username']}")

# weather_mcp_server.py
from fastmcp import FastMCP, Tool
from pydantic import BaseModel, Field
import aiohttp
import json
from typing import Optional
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize FastMCP server
mcp = FastMCP("weather-agent")

# Define request/response models with Pydantic
class WeatherForecastRequest(BaseModel):
"""Weather forecast request parameters"""
latitude: float = Field(..., description="Location latitude (-90 to 90)")
longitude: float = Field(..., description="Location longitude (-180 to 180)")
days: int = Field(default=3, description="Days to forecast (1-7)")
units: str = Field(default="metric", description="Temperature units (metric/imperial)")

class WeatherData(BaseModel):
"""Individual weather data point"""
time: str
temperature: float
condition: str
wind_speed: float
precipitation: float

class WeatherForecastResponse(BaseModel):
"""Complete weather forecast response"""
location: dict
forecast_periods: int
forecast: list[WeatherData]
updated_at: str

class AlertThreshold(BaseModel):
"""Weather alert threshold"""
condition_type: str
threshold_value: float
unit: str

class WeatherAlertsResponse(BaseModel):
"""Weather alert response"""
location: dict
active_alerts: list[str]
thresholds_exceeded: list[dict]

class ExternalWeatherAPI:
"""Handle external weather API calls"""

def __init__(self, api_key: str = None):
self.api_key = api_key or "demo" # NWS API doesn't require key
self.base_url = "https://api.weather.gov"
self.session: Optional[aiohttp.ClientSession] = None

async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()

async def get_forecast(self, latitude: float, longitude: float,
days: int = 3) -> dict:
"""Get weather forecast from National Weather Service API"""

try:
# Step 1: Get grid points for coordinates
points_url = f"{self.base_url}/points/{latitude},{longitude}"

async with self.session.get(points_url, timeout=10) as resp:
if resp.status != 200:
raise Exception(f"NWS API error: {resp.status}")

points_data = await resp.json()
forecast_url = points_data['properties']['forecast']

logger.info(f"Retrieved grid points for {latitude},{longitude}")

# Step 2: Get forecast from URL
async with self.session.get(forecast_url, timeout=10) as resp:
if resp.status != 200:
raise Exception(f"Forecast API error: {resp.status}")

forecast_data = await resp.json()

# Step 3: Process forecast periods
periods = forecast_data['properties']['periods'][:days*2]

processed_forecast = []
for period in periods:
processed_forecast.append({
"time": period['startTime'],
"temperature": period['temperature'],
"condition": period['shortForecast'],
"wind_speed": self._parse_wind_speed(period['windSpeed']),
"precipitation": 0 # NWS doesn't always provide this
})

return {
"location": {
"latitude": latitude,
"longitude": longitude,
"name": points_data['properties'].get('relativeLocation', {}).get('properties', {}).get('city')
},
"forecast_periods": len(processed_forecast),
"forecast": processed_forecast,
"updated_at": datetime.utcnow().isoformat()
}

except Exception as e:
logger.error(f"Weather API error: {str(e)}")
raise

async def get_alerts(self, latitude: float, longitude: float) -> dict:
"""Get active weather alerts for location"""

try:
alerts_url = f"{self.base_url}/alerts/active?point={latitude},{longitude}"

async with self.session.get(alerts_url, timeout=10) as resp:
if resp.status != 200:
raise Exception(f"Alerts API error: {resp.status}")

alerts_data = await resp.json()

active_alerts = []
for feature in alerts_data.get('features', []):
props = feature.get('properties', {})
active_alerts.append({
"event": props.get('event'),
"severity": props.get('severity'),
"headline": props.get('headline')
})

return {
"location": {
"latitude": latitude,
"longitude": longitude
},
"active_alerts": [a['headline'] for a in active_alerts],
"alert_count": len(active_alerts)
}

except Exception as e:
logger.error(f"Alerts API error: {str(e)}")
raise

@staticmethod
def _parse_wind_speed(wind_string: str) -> float:
"""Parse wind speed string (e.g., '10 mph' or '16 km/h')"""
import re
match = re.search(r'(\d+)', wind_string or "0")
return float(match.group(1)) if match else 0

# MCP Tool Definitions
@mcp.tool()
async def get_weather_forecast(request: WeatherForecastRequest) -> WeatherForecastResponse:
"""
Get detailed weather forecast for a specific location.

This tool retrieves multi-day weather forecasts using the National Weather Service API.
Returns temperature, conditions, wind speed, and precipitation data.

Args:
latitude: Location latitude (-90 to 90)
longitude: Location longitude (-180 to 180)
days: Number of days to forecast (1-7)
units: Temperature units (metric or imperial)

Returns:
Complete forecast with location and weather periods
"""

logger.info(f"Forecast request: ({request.latitude}, {request.longitude})")

async with ExternalWeatherAPI() as api:
forecast = await api.get_forecast(
request.latitude,
request.longitude,
request.days
)

return WeatherForecastResponse(**forecast)

@mcp.tool()
async def get_weather_alerts(latitude: float = Field(..., description="Location latitude"),
longitude: float = Field(..., description="Location longitude")) -> WeatherAlertsResponse:
"""
Get active weather alerts for a location.

Returns current severe weather alerts, watches, and warnings affecting the specified location.

Args:
latitude: Location latitude (-90 to 90)
longitude: Location longitude (-180 to 180)

Returns:
Active alerts with severity levels
"""

logger.info(f"Alerts request: ({latitude}, {longitude})")

async with ExternalWeatherAPI() as api:
alerts = await api.get_alerts(latitude, longitude)

return WeatherAlertsResponse(**alerts)

@mcp.tool()
async def analyze_weather_conditions(forecast_summary: str = Field(..., description="Summary of weather conditions")) -> dict:
"""
Analyze weather conditions and provide recommendations.

Uses Claude to interpret weather patterns and generate actionable advice.

Args:
forecast_summary: Natural language weather summary

Returns:
Analysis and recommendations
"""

# This tool demonstrates how to chain Agent Core capabilities
return {
"analysis": "Weather analysis would be performed by Agent Core",
"recommendations": [
"Monitor forecast updates regularly",
"Prepare for potential weather changes"
]
}

# Server startup
if __name__ == "__main__":
import uvicorn

logger.info("Starting Weather MCP Server")
logger.info("Available tools: get_weather_forecast, get_weather_alerts, analyze_weather_conditions")

# Run FastMCP server
uvicorn.run(
"weather_mcp_server:mcp",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info"
)

# jwt_authorizer.py
import json
import jwt
import logging
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
from functools import lru_cache
import httpx

logger = logging.getLogger(__name__)

class JWTAuthorizer:
"""Validate JWT tokens from Amazon Cognito"""

def __init__(self, user_pool_id: str, region: str = "us-east-1"):
self.user_pool_id = user_pool_id
self.region = region
self.cache = {}
self._jwks_url = f"https://cognito-idp.{region}.amazonaws.com/{user_pool_id}/.well-known/jwks.json"

async def get_jwks(self) -> Dict[str, Any]:
"""Fetch JSON Web Key Set from Cognito"""

cache_key = "jwks"

# Check cache first (valid for 1 hour)
if cache_key in self.cache:
cached_data = self.cache[cache_key]
if datetime.now() < cached_data['expires']:
return cached_data['data']

try:
async with httpx.AsyncClient() as client:
response = await client.get(self._jwks_url, timeout=5)
response.raise_for_status()

jwks = response.json()

# Cache for 1 hour
self.cache[cache_key] = {
'data': jwks,
'expires': datetime.now() + timedelta(hours=1)
}

logger.info("JWKS fetched and cached")
return jwks

except Exception as e:
logger.error(f"Failed to fetch JWKS: {str(e)}")
raise

def get_public_key(self, token: str, jwks: Dict) -> Optional[str]:
"""Extract public key from JWKS"""

try:
# Decode header without verification to get 'kid'
header = jwt.get_unverified_header(token)
kid = header.get('kid')

if not kid:
logger.error("Token header missing 'kid'")
return None

# Find matching key in JWKS
for key in jwks.get('keys', []):
if key.get('kid') == kid:
return jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(key))

logger.error(f"Key ID {kid} not found in JWKS")
return None

except Exception as e:
logger.error(f"Error extracting public key: {str(e)}")
return None

async def validate_token(self, token: str, expected_audience: Optional[str] = None) -> Dict[str, Any]:
"""Validate JWT token from Cognito"""

try:
# Remove 'Bearer ' prefix if present
if token.startswith('Bearer '):
token = token[7:]

# Get JWKS
jwks = await self.get_jwks()
public_key = self.get_public_key(token, jwks)

if not public_key:
raise ValueError("Could not extract public key")

# Decode and verify token
payload = jwt.decode(
token,
public_key,
algorithms=['RS256'],
audience=expected_audience if expected_audience else None,
options={
'verify_exp': True,
'verify_aud': bool(expected_audience)
}
)

logger.info(f"Token validated for user: {payload.get('email')}")
return {
'valid': True,
'payload': payload,
'user_id': payload.get('sub'),
'email': payload.get('email'),
'agent_role': payload.get('custom:agent_role')
}

except jwt.ExpiredSignatureError:
logger.warning("Token has expired")
return {'valid': False, 'error': 'Token expired'}
except jwt.InvalidSignatureError:
logger.warning("Invalid token signature")
return {'valid': False, 'error': 'Invalid signature'}
except Exception as e:
logger.error(f"Token validation error: {str(e)}")
return {'valid': False, 'error': str(e)}

class AgentCoreAuthMiddleware:
"""Middleware for Agent Core JWT authentication"""

def __init__(self, user_pool_id: str, region: str = "us-east-1"):
self.authorizer = JWTAuthorizer(user_pool_id, region)

async def authenticate_agent_request(self,
authorization_header: Optional[str],
expected_audience: Optional[str] = None) -> Dict[str, Any]:
"""Authenticate incoming Agent Core request"""

if not authorization_header:
return {
'authenticated': False,
'error': 'Missing Authorization header'
}

validation_result = await self.authorizer.validate_token(
authorization_header,
expected_audience
)

return {
'authenticated': validation_result.get('valid', False),
'user_id': validation_result.get('user_id'),
'email': validation_result.get('email'),
'agent_role': validation_result.get('agent_role'),
'error': validation_result.get('error')
}

# Usage in Agent Core deployment
class SecureWeatherAgent:
"""Weather agent with JWT authentication"""

def __init__(self, user_pool_id: str):
self.auth_middleware = AgentCoreAuthMiddleware(user_pool_id)

async def invoke_with_auth(self,
auth_header: str,
agent_input: Dict[str, Any]) -> Dict[str, Any]:
"""Invoke agent with authentication"""

# Authenticate request
auth_result = await self.auth_middleware.authenticate_agent_request(
auth_header,
expected_audience="weather-agent-api"
)

if not auth_result['authenticated']:
return {
'error': auth_result.get('error'),
'status_code': 401
}

# Proceed with agent invocation
logger.info(f"Authenticated request from {auth_result['email']}")

# Your agent logic here
return {
'status': 'success',
'user_id': auth_result['user_id'],
'result': await self._invoke_agent(agent_input)
}

async def _invoke_agent(self, agent_input: Dict[str, Any]) -> Dict[str, Any]:
"""Actual agent invocation logic"""
return {"message": "Agent processing..."}

# deploy_to_agent_core.py
import boto3
import json
import time
from datetime import datetime

class AgentCoreDeployment:
"""Manage Bedrock Agent Core deployments"""

def __init__(self, region: str = "us-east-1"):
self.bedrock_agent = boto3.client('bedrock-agent', region_name=region)
self.codebuild = boto3.client('codebuild', region_name=region)

def trigger_remote_build(self, project_name: str,
source_version: str = "main") -> str:
"""Trigger CodeBuild for cross-platform compilation"""

response = self.codebuild.start_build(
projectName=project_name,
sourceVersion=source_version,
environmentVariablesOverride=[
{
'name': 'TARGET_ARCHITECTURE',
'value': 'ARM64',
'type': 'PLAINTEXT'
},
{
'name': 'BUILD_TIMESTAMP',
'value': datetime.now().isoformat(),
'type': 'PLAINTEXT'
}
]
)

build_id = response['build']['id']
print(f"✓ Started CodeBuild: {build_id}")

return build_id

def wait_for_build(self, build_id: str, max_wait_minutes: int = 15) -> bool:
"""Poll build status until completion"""

start_time = time.time()
max_seconds = max_wait_minutes * 60

while (time.time() - start_time) < max_seconds:
response = self.codebuild.batch_get_builds(ids=[build_id])
build = response['builds']

current_status = build['buildStatus']

if current_status == 'SUCCEEDED':
print(f"✓ Build succeeded")
return True
elif current_status in ['FAILED', 'FAULT', 'TIMED_OUT', 'STOPPED']:
print(f"✗ Build {current_status}")
# Print logs for debugging
self._print_build_logs(build)
return False

print(f" Build status: {current_status}...")
time.sleep(10)

print(f"✗ Build did not complete within {max_wait_minutes} minutes")
return False

def deploy_agent(self, agent_config: dict,
agent_name: str) -> str:
"""Deploy agent to Bedrock Agent Core"""

try:
# Check if agent exists
existing_agents = self.bedrock_agent.list_agents()

agent_id = None
for agent in existing_agents.get('agentSummaries', []):
if agent['agentName'] == agent_name:
agent_id = agent['agentId']
break

if agent_id:
# Update existing agent
print(f"Updating existing agent: {agent_id}")
response = self.bedrock_agent.update_agent(
agentId=agent_id,
agentName=agent_name,
**agent_config
)
else:
# Create new agent
print(f"Creating new agent: {agent_name}")
response = self.bedrock_agent.create_agent(
agentName=agent_name,
**agent_config
)
agent_id = response['agent']['agentId']

print(f"✓ Agent deployment initiated: {agent_id}")

# Create agent version (enables deployment)
version_response = self.bedrock_agent.create_agent_version(
agentId=agent_id
)

version = version_response['agentVersion']['version']
print(f"✓ Created agent version: {version}")

# Prepare agent for use
self.bedrock_agent.prepare_agent(agentId=agent_id)
print(f"✓ Agent prepared for deployment")

return agent_id

except Exception as e:
print(f"✗ Deployment error: {str(e)}")
raise

def _print_build_logs(self, build: dict):
"""Print CodeBuild logs for debugging"""
logs_config = build.get('logs', {})
s3_logs = logs_config.get('s3Logs', {})

if s3_logs.get('s3LogsArn'):
print(f"Logs available at: {s3_logs['s3LogsArn']}")

# Deployment workflow
if __name__ == "__main__":
deployer = AgentCoreDeployment()

# Step 1: Trigger cross-platform build
build_id = deployer.trigger_remote_build("weather-agent-build")

# Step 2: Wait for build completion
if not deployer.wait_for_build(build_id):
print("Build failed. Check CodeBuild logs.")
exit(1)

# Step 3: Deploy to Agent Core
agent_config = {
'foundationModel': 'anthropic.claude-3-sonnet-20240229-v1:0',
'agentResourceRoleArn': 'arn:aws:iam::ACCOUNT:role/AgentCoreRole',
'description': 'Intelligent weather forecasting agent',
}

agent_id = deployer.deploy_agent(agent_config, "weather-agent")
print(f"\n✓ Deployment complete: {agent_id}")

# test_weather_agent.py
import pytest
import asyncio
import json
from weather_mcp_server import (
get_weather_forecast,
get_weather_alerts,
WeatherForecastRequest
)

class TestWeatherAgent:
"""Test weather agent MCP compliance"""

@pytest.mark.asyncio
async def test_weather_forecast_valid_coordinates(self):
"""Test forecast with valid coordinates"""

request = WeatherForecastRequest(
latitude=40.7128, # New York City
longitude=-74.0060,
days=3,
units="metric"
)

result = await get_weather_forecast(request)

# Validate MCP response structure
assert result.location is not None
assert result.forecast_periods > 0
assert len(result.forecast) == result.forecast_periods
assert result.updated_at is not None

# Validate forecast data
for period in result.forecast:
assert period.time is not None
assert -50 < period.temperature < 60 # Reasonable temp range
assert period.condition is not None
assert period.wind_speed >= 0

@pytest.mark.asyncio
async def test_weather_forecast_boundary_coordinates(self):
"""Test forecast at coordinate boundaries"""

test_cases = [
(90, 0), # North pole
(-90, 0), # South pole
(0, 180), # Date line
(0, -180), # Date line opposite
]

for lat, lon in test_cases:
request = WeatherForecastRequest(
latitude=lat,
longitude=lon,
days=1
)

# Should not raise exception
try:
result = await get_weather_forecast(request)
assert result is not None
except Exception as e:
# Some locations may not have forecast data
assert "not found" in str(e).lower() or "no data" in str(e).lower()

@pytest.mark.asyncio
async def test_weather_alerts_active(self):
"""Test alert retrieval"""

result = await get_weather_alerts(
latitude=40.7128,
longitude=-74.0060
)

# Validate MCP response structure
assert result.location is not None
assert isinstance(result.active_alerts, list)
assert result.alert_count >= 0

@pytest.mark.asyncio
async def test_jwt_authentication(self):
"""Test JWT token validation"""

from jwt_authorizer import JWTAuthorizer

authorizer = JWTAuthorizer("us-east-1_POOLID")

# Test with invalid token
result = await authorizer.validate_token("invalid.token.format")
assert result['valid'] == False

@pytest.mark.asyncio
async def test_concurrent_requests(self):
"""Test agent handles concurrent MCP requests"""

requests = [
WeatherForecastRequest(latitude=40.7128, longitude=-74.0060),
WeatherForecastRequest(latitude=51.5074, longitude=-0.1278), # London
WeatherForecastRequest(latitude=48.8566, longitude=2.3522), # Paris
]

# Execute concurrently
results = await asyncio.gather(*[
get_weather_forecast(req) for req in requests
])

assert len(results) == 3
assert all(r.forecast_periods > 0 for r in results)

# Run tests
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])