Notamify Python SDK

The Notamify Python SDK is an open-source library that streamlines NOTAM data fetching, listener management, and real-time webhook handling for the Notamify API v2 and Watcher API.

The Notamify Python SDK is an open-source, MIT-licensed Python client that wraps both the Notamify API v2 and the Notamify Watcher API into a single, typed interface. It is designed to streamline NOTAM data fetching, eliminate boilerplate HTTP code, and make the Watcher API accessible with just a few lines of Python.

Key highlights:

• Single NotamifyClient for both NOTAM queries and Watcher listener management

• Fully typed with Pydantic v2 models

• Auto-paginating iterators for NOTAM endpoints

• Built-in webhook signature verification

• Embedded webhook receiver for local development and testing

• Cloudflare tunnel integration for exposing local endpoints

• Requires Python 3.10+

Installation

pip install notamify-sdk

Or with uv:

uv add notamify-sdk

Authentication

The SDK supports multiple ways to provide your API token:

from notamify_sdk import NotamifyClient, ConfigStore

# Option 1: Pass the token directly
client = NotamifyClient(token="YOUR_API_KEY")

# Option 2: Load from environment variable or config file
cfg = ConfigStore().load()
client = NotamifyClient(token=cfg.token)

Fetching NOTAMs

The SDK provides typed query models for each endpoint and two ways to fetch data: single-page methods and auto-paginating iterators.

Active NOTAMs

from datetime import datetime, timedelta
from notamify_sdk import NotamifyClient, ActiveNotamsQuery

client = NotamifyClient(token="YOUR_API_KEY")

query = ActiveNotamsQuery(
    location=["KJFK", "KLAX"],
    starts_at=datetime.now(),
    ends_at=datetime.now() + timedelta(days=1)
)

# Auto-paginated — iterates all pages automatically
for notam in client.notams.active(query):
    print(f"{notam.notam_number} [{notam.interpretation.excerpt}]")

Nearby NOTAMs

from notamify_sdk import NotamifyClient, NearbyNotamsQuery

client = NotamifyClient(token="YOUR_API_KEY")

query = NearbyNotamsQuery(lat=51.4775, lon=-0.4614, radius_nm=15.0)

for notam in client.notams.nearby(query):
    print(f"{notam.notam_number}: {notam.interpretation.excerpt}")

Raw NOTAMs

Raw NOTAMs are returned without AI interpretation — useful for prefetching before pulling interpretations:

from notamify_sdk import NotamifyClient, ActiveNotamsQuery

client = NotamifyClient(token="YOUR_API_KEY")

result = client.get_raw_notams(ActiveNotamsQuery(location=["EDDF"]))

for notam in result.notams:
    print(notam.icao_message)

Historical NOTAMs

from datetime import date
from notamify_sdk import NotamifyClient, HistoricalNotamsQuery

client = NotamifyClient(token="YOUR_API_KEY")

query = HistoricalNotamsQuery(
    location=["EDDM"],
    valid_at=date(2024, 9, 20),
)

for notam in client.notams.historical(query):
    print(f"{notam.notam_number}: {notam.message}")

---

Async Briefings

The SDK wraps the two-step async briefing workflow:

import time
from datetime import datetime, timedelta
from notamify_sdk import (
    NotamifyClient,
    GenerateFlightBriefingRequest,
    LocationWithType,
)

client = NotamifyClient(token="YOUR_API_KEY")

job = client.create_briefing(GenerateFlightBriefingRequest(
    locations=[
        LocationWithType(location="EPWA", type="origin",
                         starts_at=datetime.now() + timedelta(hours=3),
                         ends_at=datetime.now() + timedelta(hours=3)),
        LocationWithType(location="EGLL", type="destination",
                         starts_at=datetime.now() + timedelta(hours=6),
                         ends_at=datetime.now() + timedelta(hours=8)),
    ],
    aircraft_type="B738",
    origin_runway="RWY11",
    destination_runway="RWY27L",
))


print(f"Job submitted: {job.uuid}")

# Poll for completion
while True:
    status = client.get_briefing_status(job.uuid)
    if status.status == "completed":
        briefing = status.response
        break
    elif status.status == "failed":
        raise RuntimeError("Briefing generation failed")
    time.sleep(2)

---

Watcher API: Real-Time NOTAM Monitoring

Instead of polling for NOTAMs, Watcher API allows you to register listeners with filters, and Notamify delivers matching NOTAMs to your webhook URL in real time.

The SDK makes this entire workflow: creating listeners, receiving webhooks, verifying signatures.

Managing Listeners

from notamify_sdk import NotamifyClient

client = NotamifyClient(token="YOUR_API_KEY")

# List existing listeners
for listener in client.list_listeners():
    print(f"{listener.name}: {listener.id} (active={listener.active})")

# Create a new listener
listener = client.create_listener(
    name="European Hub Monitor",
    webhook_url="https://your-server.com/webhooks/notamify",
    filters={
        "notam_icao": ["EDDF", "EDDM", "EGLL", "LFPG"],
        "category": ["RUNWAY", "OBSTACLES", "AIRSPACE"],
    },
    lifecycle_enabled=True,
    mode="prod",
)
print(f"Created: {listener.id}")

# Update filters
client.update_listener(
    listener.id,
    filters={
        "notam_icao": ["EDDF", "EDDM", "EGLL", "LFPG", "EHAM"],
        "category": ["all"],
    },
)

# Pause a listener
client.update_listener(listener.id, active=False)

# Delete a listener
client.delete_listener(listener.id)

Listener Filters

Filters control which NOTAMs are delivered to your webhook. Filters use OR logic within a field and AND logic across fields:

from notamify_sdk import ListenerFilters

filters = ListenerFilters(
    notam_icao=["KJFK", "KEWR", "KLGA"],         # OR — any of these airports
    category=["OBSTACLES"],                      # OR — either category
    subcategory=["CRANE_OPERATIONS"],            # AND with category
    affected_element=[
        {"type": "RUNWAY", "effect": "RESTRICTED"} # Match restricted Runway
    ],
)

listener = client.create_listener(
    name="NYC Area Runway & Obstacles",
    webhook_url="https://your-server.com/webhooks",
    filters=filters,
    mode="prod",
)

Lifecycle Events

Enable lifecycle tracking to receive notifications when NOTAMs you've already been delivered are cancelled or replaced:

listener = client.create_listener(
    name="EDDM Full Lifecycle",
    webhook_url="https://your-server.com/webhooks",
    filters={"notam_icao": ["EDDM"]},
    lifecycle_enabled=True,
    mode="prod",
)

When a previously delivered NOTAM is replaced or cancelled, Watcher sends a lifecycle event with a change object pointing to the original NOTAM. Lifecycle events are free — they do not consume credits.

Sandbox Testing

Test your webhook integration without waiting for real NOTAMs:

# Switch a listener to sandbox mode
sandbox_listener = client.create_listener(
    name="Test Listener",
    webhook_url="https://your-server.com/webhooks",
    filters={"notam_icao": ["KJFK"]},
    mode="sandbox",
)

# Trigger a sandbox delivery with a specific NOTAM ID
result = client.send_sandbox_message(
    sandbox_listener.id,
    notam_id="some-notam-uuid",
)
print(f"Sandbox delivery: {result}")

Webhook Secret Management

# View your current (masked) webhook secret
masked = client.get_webhook_secret_masked()
print(f"Current secret: {masked}")

# Rotate the secret — previous key stays active for 3 hours
new_secret = client.rotate_webhook_secret()
print(f"New secret: {new_secret}")

---

Receiving Webhooks

The SDK includes two tools for handling incoming webhook deliveries: signature verification and an embedded receiver for local development.

Signature Verification

Every webhook request from Notamify includes an X-Notamify-Signature header. Use the SDK to verify it:

from notamify_sdk import verify_signature, SignatureVerificationError

def handle_webhook(request):
    try:
        verify_signature(
            header=request.headers["X-Notamify-Signature"],
            secret="nmf_wh_your_secret",
            body=request.body,
            tolerance_seconds=600,
        )
    except SignatureVerificationError as e:
        return {"error": str(e)}, 401

    # Signature valid — parse the event
    event = WatcherWebhookEvent.model_validate_json(request.body)
    process_event(event)
    return {"ok": True}, 200

Parsing Webhook Events

from notamify_sdk import WatcherWebhookEvent

event = WatcherWebhookEvent.model_validate_json(raw_body)

if event.kind == "interpretation":
    notam = event.notam
    print(f"New NOTAM: {notam.notam_number}")
    print(f"  Airport: {notam.icao_code}")
    print(f"  Category: {notam.interpretation.category}")
    print(f"  Description: {notam.interpretation.description}")

    # Access map elements for GeoJSON geometry
    for element in (notam.interpretation.map_elements or []):
        print(f"  Map element: {element.element_type}")
        if element.geojson:
            print(f"  GeoJSON: {element.geojson}")

elif event.kind == "lifecycle":
    print(f"Lifecycle event for NOTAM: {event.notam.notam_number}")
    print(f"  Change type: {event.change.notam_type}")  # "C" or "R"
    print(f"  Original NOTAM ID: {event.change.changed_notam_id}")

# Access airport context metadata
if event.context:
    loc = event.context.location
    print(f"  Location: {loc.name} ({loc.icao}), {loc.iso_country_name}")
    print(f"  Coordinates: {loc.coordinates.lat}, {loc.coordinates.lon}")

Local Development with Cloudflare Tunnels

To receive webhooks on your local machine, the SDK integrates with cloudflared to create a public tunnel:

import time

from notamify_sdk import (
    APIError,
    CloudflaredManager,
    NotamifyClient,
    ReceiverConfig,
    WebhookReceiver,
)

receiver = WebhookReceiver(
    ReceiverConfig(
        host="127.0.0.1",
        port=8080,
        path="/webhooks/notamify",
        allow_unsigned_dev=True,
    ),
    on_event=lambda e: print(e.parse_webhook_event()),
)
receiver.start()

tunnel = CloudflaredManager(local_url="http://127.0.0.1:8080")
info = tunnel.start(timeout_seconds=20)
print(f"Public URL: {info.public_url}")

client = NotamifyClient(token="YOUR_API_KEY")
listener = client.create_listener(
    name="Local Dev",
    webhook_url=f"{info.public_url}/webhooks/notamify",
    filters={"notam_icao": ["KJFK"]},
    mode="sandbox",
    lifecycle={"enabled": False},
)

try:
    # Fresh trycloudflare.com quick tunnels can need a short DNS propagation delay,
    # so retry server-side (5xx) delivery errors for a short while.
    for attempt in range(15):
        try:
            client.send_sandbox_message(listener.id)
            break
        except APIError as exc:
            if exc.status < 500 or attempt == 14:
                raise
            time.sleep(4)
finally:
    client.delete_listener(listener.id)
    tunnel.stop()
    receiver.stop()

Note: Requires the cloudflared binary in your PATH. Install it from Cloudflare's downloads page.

Please also note that Cloudflared itself might require time to propagate.

---

Error Handling

The SDK raises typed exceptions for all error conditions:

from notamify_sdk import NotamifyClient, APIError

client = NotamifyClient(token="YOUR_API_KEY")

try:
    result = client.get_active_notams({"location": ["KJFK"]})
except APIError as e:
    print(f"HTTP {e.status}: {e.message}")
    # e.status  — HTTP status code (0 for connection errors)
    # e.message — error description
    # e.payload — raw error payload from the API

Exception	When
APIError	Any non-2xx HTTP response or connection failure
SignatureVerificationError	Webhook signature mismatch, expired timestamp, or missing header
CloudflaredError	cloudflared binary missing, tunnel creation failure, or timeout
pydantic.ValidationError	Invalid query parameters (e.g., per_page > 30, page < 1)

Complete Example: Production Watcher Service

A minimal production service that listens for NOTAM events and processes them:

from notamify_sdk import (
    NotamifyClient,
    WebhookReceiver,
    ReceiverConfig,
    WatcherWebhookEvent,
    ConfigStore,
)

# Load config from environment
cfg = ConfigStore().load()
client = NotamifyClient(token=cfg.token)

def handle_event(raw_event):
    event = raw_event.parse_webhook_event()

    if event.kind == "interpretation":
        notam = event.notam
        interp = notam.interpretation

        print(f"[NEW] {notam.notam_number} at {notam.icao_code}")
        print(f"  Category: {interp.category}/{interp.subcategory}")
        print(f"  {interp.description}")

        for elem in interp.affected_elements:
            print(f"  Affected: {elem.type} {elem.identifier} ({elem.effect})")

    elif event.kind == "lifecycle":
        change_type = "REPLACED" if event.change.notam_type == "R" else "CANCELLED"
        print(f"[{change_type}] {event.notam.notam_number}")
        print(f"  Original NOTAM: {event.change.changed_notam_id}")

# Ensure listener exists
listeners = client.list_listeners()
if not any(l.name == "Production Service" for l in listeners):
    client.create_listener(
        name="Production Service",
        webhook_url="https://your-server.com/webhooks/notamify",
        filters={"notam_icao": ["KJFK", "KLAX", "EGLL"], "category": ["all"]},
        lifecycle_enabled=True,
        mode="prod",
    )

# Start receiver
receiver = WebhookReceiver(
    ReceiverConfig(
        host="0.0.0.0",
        port=8080,
        path="/webhooks/notamify",
        secret=cfg.webhook_secret,
        require_signature=True,
    ),
    on_event=handle_event,
)
receiver.start()
print("Watcher service running on :8080")

# Keep alive
import signal
signal.pause()

Resources

• Source code: github.com/skymerse/notamify-sdk-python

• API Manager: notamify.com/api-manager

• Watcher API docs: NOTAM Watcher API

• Webhook messages: Watcher API Webhook messages

• Webhook security: Webhook Security