Send a Push Notification From a LangChain Agent

A retrieval chain over a few thousand documents does not fail loudly. It sits there. Embedding calls go out, rate limits push back, retries happen, and the whole thing takes eleven minutes instead of the two you assumed. You find out by tabbing back.

LangChain already emits an event at the exact moment that chain ends. The callback system exists precisely so you can hang something off it without editing the chain. So hang a push notification off it.

Why a callback and not a wrapper

You could wrap the .invoke() call in a try/finally and be done in five lines. That works, and for a single script it is the right amount of engineering.

The callback earns its place when the chain is not yours to wrap: it lives behind a FastAPI route, or inside LangGraph, or three layers down in someone else's abstraction. A handler attaches at call time and fires wherever the chain actually runs, which is the case that made this worth writing up.

Step 1: The app key

Create an application in TheNotificationApp, copy the app_key, and put it in the environment. One header, no OAuth, no token refresh:

export TNA_APP_KEY="your_app_key_here"

Step 2: The handler

import os, time, requests
from typing import Any
from langchain_core.callbacks import BaseCallbackHandler

TNA_URL = "https://thenotification.app/api/sendNotification"

def _push(title: str, body: str, link: str | None = None) -> None:
    try:
        requests.post(
            TNA_URL,
            headers={"app_key": os.environ["TNA_APP_KEY"], "Content-Type": "application/json"},
            json={"title": title, "body": body, **({"link": link} if link else {})},
            timeout=10,
        )
    except Exception as exc:
        print(f"[notify] {exc}")

class PhoneNotifier(BaseCallbackHandler):
    """Push a notification when the outermost chain finishes or blows up."""

    def __init__(self, label: str = "Chain"):
        self.label = label
        self.started = None

    def on_chain_start(self, serialized, inputs, *, run_id=None, parent_run_id=None, **kw):
        if parent_run_id is None:
            self.started = time.monotonic()

    def on_chain_end(self, outputs: dict[str, Any], *, run_id=None, parent_run_id=None, **kw):
        if parent_run_id is not None:
            return
        elapsed = int(time.monotonic() - (self.started or time.monotonic()))
        text = str(outputs.get("output") or outputs)[:180].replace("\n", " ")
        _push(f"{self.label} finished in {elapsed}s", text)

    def on_chain_error(self, error: BaseException, *, parent_run_id=None, **kw):
        if parent_run_id is not None:
            return
        _push(f"{self.label} failed", f"{type(error).__name__}: {error}"[:180])

The parent_run_id is None guard is the important line. LangChain fires these events for every nested runnable, so without it a chain of six steps sends six notifications and you learn to ignore all of them. Only the outermost run has no parent.

Step 3: Attach it

Nothing about the chain changes. The handler goes in at call time:

chain = prompt | llm | parser

result = chain.invoke(
    {"question": "Summarise every incident from last week."},
    config={"callbacks": [PhoneNotifier(label="Incident summary")]},
)

For an agent executor it is the same argument, and for LangGraph it goes in the same config dict. That is the payoff of using the callback system rather than a wrapper: one handler covers all three.

Step 4: Make the notification worth reading

"Chain finished" tells you to go look. Put the actual answer in the body and you can often skip going to look at all, which is the entire point:

_push(
    f"Incident summary ready ({elapsed}s)",
    text[:180],
    link="https://your-dashboard.example.com/incidents",
)

A tapped notification opens the link. Point it at the dashboard, the run in LangSmith, or wherever the output landed. The full field list, including image, is in the API reference.

Agents, not just chains

An agent executor fires the same chain events, so the handler above already works. What it does not give you is any sense of what the agent did on the way, and for a run that takes twenty minutes you sometimes want the shape of it.

Two more callbacks cover that without turning your phone into a log tail. Count the tools instead of announcing them:

class AgentNotifier(PhoneNotifier):
    def __init__(self, label: str = "Agent"):
        super().__init__(label)
        self.tools: list[str] = []

    def on_tool_start(self, serialized, input_str, **kw):
        self.tools.append(serialized.get("name", "tool"))

    def on_chain_end(self, outputs, *, parent_run_id=None, **kw):
        if parent_run_id is not None:
            return
        elapsed = int(time.monotonic() - (self.started or time.monotonic()))
        text = str(outputs.get("output") or outputs)[:140].replace("\n", " ")
        _push(
            f"{self.label} done in {elapsed}s",
            f"{len(self.tools)} tool calls. {text}",
        )

Now the notification reads "Agent done in 340s: 14 tool calls. Found 3 incidents matching the pattern." That is enough to judge whether the run went the way you expected without opening anything.

Running it behind an API

The case that made the callback worth the extra code: the chain is invoked by a request handler, and you are not the one calling it.

from fastapi import FastAPI

app = FastAPI()

@app.post("/summarise")
async def summarise(payload: dict):
    return await chain.ainvoke(
        payload,
        config={"callbacks": [PhoneNotifier(label="Summary API")]},
    )

The async methods have the same names, so ainvoke works with the same handler. Worth being deliberate here though: a handler attached to a public endpoint fires once per request, and one notification per request is a pager, not an alert. Attach it to the slow internal jobs, not the hot path.

The failure case is the one that pays for itself

A chain that finishes eleven minutes late is annoying. A chain that dies at minute two and sits there dead until you check is expensive, because you lose the whole window you thought was productive.

on_chain_error gives you that for free, and the exception type in the title is enough to know whether it needs you now or after coffee. A RateLimitError can wait. A ValidationError in your parser means the last eleven minutes produced nothing and the sooner you restart the better.

Watch the quota if the chain is chatty

The free tier is 100 notifications for the lifetime of the account, not per month. One notification per outer chain run is a slow burn and that lasts months. Drop the parent_run_id guard by accident and a single agent run can spend twenty of them in a minute.

If you do go over, the API does not fail silently. It returns a 429 with the limit and how many you have sent, which the handler prints. Pro lifts it to 1,000 a month for $2.99. Either way, notify on the run, not on the step.

Where to take it

The same handler shape works for anything with a lifecycle: a training loop, a batch job, a nightly ETL. If you are running agents rather than chains, the pattern for long unattended runs is written up in notifying your phone when an agent run finishes.

Grab a free key at thenotification.app and attach the handler to the next chain you would otherwise sit and watch.

New to this? Start with what an MCP server is.