Skip to main content

​
resq_mcp.hce.tools

MCP tool wrappers for HCE domain.

​
logging

​
time

​
UTC

​
datetime

​
FastMCPError

​
IncidentValidation

​
MissionParameters

​
MAX_INCIDENTS

​
MAX_MISSIONS

​
incidents

​
mcp

​
missions

​
logger

​
validate_incident

@mcp.tool()
async def validate_incident(val: IncidentValidation) -> str
Submit validation result for an incident report. Used by human operators or automated validation systems (HCE) to confirm or reject incident reports before triggering full response. Arguments:
  • val - IncidentValidation with:
    • incident_id: ID of incident being validated
    • is_confirmed: True=confirmed, False=rejected/false positive
    • validation_source: Who/what validated (e.g., “Human-Operator”)
    • correlated_pre_alert_id: Optional linked PDIE alert
    • notes: Validation reasoning and evidence
Returns:
  • str - Confirmation message indicating action taken: “Incident {id} successfully CONFIRMED.” or “Incident {id} successfully REJECTED.”
Example:
from resq_mcp.hce.models import IncidentValidation validation = IncidentValidation( … incident_id=“INC-123”, … is_confirmed=True, … validation_source=“Human-Operator-Alice”, … notes=“Confirmed via video evidence and ground reports” … ) result = await validate_incident(validation) print(result) # “Incident INC-123 successfully CONFIRMED.”
Workflow:
  1. Edge AI detects incident (low confidence)
  2. HCE cross-references with PDIE/sensors
  3. If ambiguous -> human review required
  4. Operator submits validation via this tool
  5. If confirmed -> trigger response strategy
  6. If rejected -> log as false positive, update ML model
Audit Trail: All validations logged with timestamp, source, and reasoning for post-incident analysis and ML model refinement.

​
update_mission_params

@mcp.tool()
async def update_mission_params(drone_id: str,
                                strategy_id: str,
                                is_urgent: bool = False) -> MissionParameters
Push authorized mission parameters to a drone for an approved strategy. Completes the deployment workflow after a strategy has been approved: get_deployment_strategy -> (human approval) -> update_mission_params -> drone executes. Arguments:
  • drone_id - Target drone identifier (e.g., “DRONE-Alpha”).
  • strategy_id - Approved strategy ID from get_deployment_strategy (e.g., “STRAT-X1Y2Z3”).
  • is_urgent - If True, sets risk_tolerance=0.9 (aggressive routing) instead of the default 0.5. Must be explicitly set — urgency is never derived from the strategy_id string to prevent injection attacks.
Returns:
  • MissionParameters - Authorized parameter set including mission ID, allowed actions, risk tolerance, and a deterministic blockchain-anchored strategy hash.
Raises:
  • FastMCPError - If the drone already has an active mission for a different strategy (conflict guard), or if the mission store is at capacity.
Example:
params = await update_mission_params(“DRONE-Alpha”, “STRAT-ABCD1234”, is_urgent=True) print(params.authorized_actions) print(params.strategy_hash) # 0xSHA256(strategy_id:mission_id)