navflow-ai 0.2.1

Python SDK for NavFlow — the data layer for AI agents.

navflow-ai

Python SDK for NavFlow — the data layer for AI agents.

Use it to:

• Send agent output back to NavFlow for routing to your sinks (Slack, webhooks, etc.)
• Parse the payload NavFlow POSTs to your agent — typed access to the trigger, context window, and metadata

Installation

pip install navflow-ai

For the typed payload parsers (requires Pydantic), install the parse extra:

pip install 'navflow-ai[parse]'

Sending agent output

from navflow_ai import NavFlow

nf = NavFlow(api_key="nf_...")  # endpoint defaults to https://ingest.navflow.ai

nf.send_output(
    payload={
        "summary":     "Auth failures spiked at 14:02",
        "severity":    "critical",
        "remediation": "Roll back deploy a3f2c91",
    },
    request_id=x_request_id,    # match the X-Request-ID header from the invocation
)

Org-scoped API keys

Project keys (nf_...) only need the key. Org keys can target any project in the org and require an explicit project_id:

nf = NavFlow(api_key="nf_org_key")
nf.send_output(
    payload={"x": 1},
    request_id=x_request_id,
    project_id=x_project_id,    # required for org-scoped keys
)

Custom endpoint / timeout

nf = NavFlow(
    api_key="nf_...",
    endpoint="https://your-custom-receiver.example.com",
    timeout=10.0,
)

Parsing the incoming payload

NavFlow POSTs a unified shape to your agent — trigger event, optional context window, metadata. The payload module gives you typed access without hand-rolling body.get(...) chains.

from fastapi import FastAPI, Header, Request
from navflow_ai import NavFlow
from navflow_ai.payload import AgentPayload

app = FastAPI()
nf  = NavFlow(api_key="nf_...")

@app.post("/process")
async def process(
    request: Request,
    x_request_id: str = Header(default=""),
    x_project_id: str = Header(default=""),
):
    payload = AgentPayload.from_dict(await request.json())

    trigger = payload.trigger
    context = payload.context     # None when context windows are off

    if context:
        # context.events is oldest-first; context.stats has count + duration
        ...

    nf.send_output(
        payload={"summary": "..."},
        request_id=x_request_id,
        project_id=x_project_id,
    )
    return {"status": "ok"}

Available models

from navflow_ai.payload import (
    AgentPayload,   # top-level: trigger, pending_triggers, context, metadata
    Context,        # key, events, stats
    WindowEvent,    # data, timestamp
    WindowStats,    # count, duration_ms, first_at, last_at
    Metadata,       # request_id, project_id, triggered_at, group_key, ...
)

All models accept extra fields gracefully (server-added fields are kept on model_extra instead of dropped).

Without the SDK

If you don't want a Python dependency, post directly to the receiver:

curl -X POST https://ingest.navflow.ai/internal/agent-output \
  -H "X-API-Key: nf_..." \
  -H "Content-Type: application/json" \
  -H "X-Project-ID: <project-uuid>"  `# only for org-scoped keys` \
  -d '{
    "request_id": "<X-Request-ID from the invocation>",
    "payload":    { "summary": "..." }
  }'

The wire shape is {"request_id": "...", "payload": <anything JSON-serializable>}.

API reference

NavFlow(api_key, endpoint="https://ingest.navflow.ai", timeout=30.0)

Construct a client. api_key may be project-scoped or org-scoped. endpoint defaults to NavFlow's hosted receiver. timeout is per-request in seconds.

nf.send_output(payload, request_id=None, project_id=None) -> dict

POST agent output to the receiver. Returns the receiver's JSON response (typically {"status": "ok"}). Raises requests.HTTPError on a non-2xx response.

• payload — any JSON-serializable value
• request_id — auto-generated UUID if omitted; pass through the invocation's X-Request-ID header for end-to-end tracing
• project_id — required when the API key is org-scoped

AgentPayload.from_dict(body) / .from_bytes(body)

Parse a request body into a typed AgentPayload. Both class methods accept the wire shape directly (_metadata is mapped to .metadata).

Compatibility

• Python 3.10+
• Drop-in alias GlassFlow = NavFlow is preserved for the legacy package name

Links

• Documentation: https://docs.navflow.ai
• Source: https://github.com/glassflow/navflow-sdk-python
• Issues: https://github.com/glassflow/navflow-sdk-python/issues
• Contact: [e-mail]

License

Apache 2.0