> For the complete documentation index, see [llms.txt](https://skymerse.gitbook.io/notamify-api/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://skymerse.gitbook.io/notamify-api/sdk/python.md).

# Notamify Python SDK

The [Notamify Python SDK](https://github.com/skymerse/notamify-sdk-python) 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

```bash
pip install notamify-sdk
```

Or with [uv](https://docs.astral.sh/uv/):

```bash
uv add notamify-sdk
```

### Authentication

The SDK supports multiple ways to provide your API token:

```python
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

<pre class="language-python"><code class="lang-python">from datetime import datetime, timedelta
from notamify_sdk import NotamifyClient, ActiveNotamsQuery

<strong>client = NotamifyClient(token="YOUR_API_KEY")
</strong>
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}]")
</code></pre>

### Nearby NOTAMs

```python
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:

```python
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

```python
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:

```python
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](/notamify-api/notam-watcher/notam-watcher-api.md) 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

```python
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:

<pre class="language-python"><code class="lang-python">from notamify_sdk import ListenerFilters

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

listener = client.create_listener(
    name="NYC Area Runway &#x26; Obstacles",
    webhook_url="https://your-server.com/webhooks",
    filters=filters,
    mode="prod",
)
</code></pre>

### Lifecycle Events

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

<pre class="language-python"><code class="lang-python">listener = client.create_listener(
    name="EDDM Full Lifecycle",
    webhook_url="https://your-server.com/webhooks",
    filters={"notam_icao": ["EDDM"]},
<strong>    lifecycle_enabled=True,
</strong>    mode="prod",
)
</code></pre>

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:

```python
# 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

```python
# 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:

```python
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

```python
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:

```python
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.
    for attempt in range(15):
        try:
            client.send_sandbox_message(listener.id)
            break
        except APIError as exc:
            msg = (exc.message or "").lower()
            retryable_dns_error = (
                exc.status >= 500
                and (
                    "no such host" in msg
                    or "temporary failure in name resolution" in msg
                    or ("dial tcp" in msg and "lookup" in msg)
                )
            )
            if not retryable_dns_error 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](https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/downloads/).
>
> Please also note that Cloudflared itself might require time to propagate.

***

## Error Handling

The SDK raises typed exceptions for all error conditions:

```python
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:

```python
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](https://github.com/skymerse/notamify-sdk-python)
* **API Manager:** [notamify.com/api-manager](https://notamify.com/api-manager)
* **Watcher API docs:** [NOTAM Watcher API](/notamify-api/notam-watcher/notam-watcher-api.md)
* **Webhook messages:** [Watcher API Webhook messages](/notamify-api/notam-watcher/watcher-api-webhook-messages.md)
* **Webhook security:** [Webhook Security](/notamify-api/notam-watcher/webhook-security.md)