# 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)


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://skymerse.gitbook.io/notamify-api/sdk/python.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.