Skip to main content

​
resq_mcp.server

ResQ MCP Server - Model Context Protocol server for disaster response coordination. This module provides the main FastMCP server implementation for ResQ, offering:
  • Simulation management via resources and tools
  • Drone fleet status and deployment
  • Incident validation and response planning
The server uses a lifespan context manager to manage background tasks for simulation processing and notification delivery.

​
asyncio

​
contextlib

​
logging

​
time

​
asynccontextmanager

​
TYPE_CHECKING

​
Any

​
FastMCP

​
settings

​
validate_environment

​
setup_telemetry

​
logger

​
MAX_SIMULATIONS

​
MAX_INCIDENTS

​
MAX_MISSIONS

active missions per session

​
COMPLETED_TTL_SECONDS

evict completed sims after 5 minutes

​
FAILED_TTL_SECONDS

evict failed sims sooner

​
INCIDENT_TTL_SECONDS

evict rejected incident records after 1 hour

​
CONFIRMED_INCIDENT_TTL_SECONDS

confirmed incidents retained for 24h

​
MISSION_TTL_SECONDS

evict stale mission records after 2h

​
simulations

​
incidents

​
missions

keyed by drone_id

​
lifespan

@asynccontextmanager
async def lifespan(server: FastMCP) -> "AsyncGenerator[None, None]"
Lifespan context manager for the MCP server with background tasks. Manages the lifecycle of background processing tasks that run for the duration of the server. Ensures clean startup and shutdown with proper task cancellation and resource cleanup. Background Tasks Started:
  • simulation_processor: Mock simulation state machine that transitions simulations from pending -> processing -> completed and sends SSE notifications to subscribed clients.
Lifecycle:
  1. Startup: Log initialization, create background tasks
  2. Running: Yield control to FastMCP server
  3. Shutdown: Cancel tasks, suppress CancelledError, log shutdown
Arguments:
  • server - The FastMCP server instance for notification dispatch.
Yields:
  • None - Control returns to FastMCP for request handling.
Notes: In production, background tasks would interface with actual simulation clusters, message queues (Redis/RabbitMQ), and maintain persistent connections to drone telemetry streams.

​
mcp

​
simulation_processor

async def simulation_processor(server: FastMCP) -> None
Background processor for simulation state transitions and notifications. Polls for pending simulations every 2 s and spawns an independent async task per simulation so that no single job’s delay blocks the others. State Machine: pending -> processing (immediate, 50% progress, task spawned) processing -> completed (after 3 s inside _process_simulation) processing -> failed (on server shutdown mid-run) Notifications: SSE resource update notifications are sent on each state transition. Notes: Production would replace this with a real job queue integration (Celery / RQ) and Unity/Unreal Engine status polling.

​
tools

​
tools

​
prompts

​
resources

​
main

def main() -> None
Console script entry point for the ResQ MCP server.