Futuristic Risk Intelligence

Instructions

Implementation Reference

• mcp_server/server.py:186-195 (handler)

  The handler function for get_maritime_traffic that loads data from maritime.json and filters by zone_id if provided.

  def handle_get_maritime_traffic(args: dict) -> str:
      data = _load_json("maritime.json")
      if "error" in data:
          return json.dumps(data)
  
      zone_id = args.get("zone_id")
      if zone_id:
          data["zones"] = [z for z in data.get("zones", []) if z.get("zone_id") == zone_id]
  
      return json.dumps(data, indent=2)

• mcp_server/server.py:132-151 (schema)

  The tool definition schema for get_maritime_traffic.

      "name": "get_maritime_traffic",
      "description": (
          "Get vessel counts in critical maritime chokepoints: Strait of Hormuz, "
          "Black Sea, Taiwan Strait, Arabian Sea, Eastern Mediterranean, Caribbean. "
          "Includes breakdown by vessel type (tanker, cargo, military, other). "
          "Data is a snapshot from AIS receivers (not full-day throughput). "
          "Military vessels often turn off AIS transponders so counts may underestimate. "
          "Updated daily. Use this to monitor supply chain disruption risks."
      ),
      "inputSchema": {
          "type": "object",
          "properties": {
              "zone_id": {
                  "type": "string",
                  "description": "Optional: filter to a single maritime zone.",
              },
          },
          "required": [],
      },
  },

• mcp_server/server.py:201-201 (registration)

  Registration of the get_maritime_traffic handler in the TOOL_HANDLERS dictionary.

  "get_maritime_traffic": handle_get_maritime_traffic,