langchain agent email callback notification: what actually works in production

Your LangChain agent just finished a 45-second research task. It synthesized three sources and handed you a clean summary. But you didn't know any of that happened, because nobody told you. No email. No ping. Nothing.

That's the gap callbacks are supposed to fill. LangChain's callback system lets you hook into agent lifecycle events (when a tool runs, when the LLM responds, when a chain completes, when the agent finishes) and trigger side effects like sending an email notification. The concept is straightforward. Getting it to work reliably in production is where things fall apart.

How to send an email notification from a LangChain agent callback#

Here's the pattern in seven steps:

1. Import BaseCallbackHandler from langchain_core.callbacks.
2. Create a subclass and override on_agent_finish with your logic.
3. Add email dispatch inside that method using SMTP or an email API.
4. Instantiate the handler and pass it to AgentExecutor via callbacks=[handler].
5. Override on_tool_error to send separate alerts when a tool fails.
6. For async agents, subclass AsyncCallbackHandler and use await in overrides.
7. Test with a mock email transport before connecting to a live provider.

The rest of this article explains each piece, then covers the production problems that tutorials skip.

What LangChain callback handlers actually do#

A callback handler is a class that listens to events during an agent's execution. LangChain fires events at specific moments: when the LLM starts generating, when a tool is called, when a chain completes, when the agent decides on an action, and when it returns its final answer.

The events most useful for email notifications are:

• on_agent_finish fires when the agent completes and returns a final answer
• on_tool_error fires when a tool throws an exception
• on_chain_end fires when any chain (including nested ones) completes
• on_llm_error fires when the language model call itself fails

Here's a minimal callback handler that sends an email when your agent finishes:

from langchain_core.callbacks import BaseCallbackHandler
from langchain.schema import AgentFinish
import smtplib
from email.mime.text import MIMEText

class EmailNotificationHandler(BaseCallbackHandler):
    def on_agent_finish(self, finish: AgentFinish, **kwargs):
        msg = MIMEText(f"Agent completed.\n\nResult: {finish.return_values}")
        msg["Subject"] = "Agent task completed"
        msg["From"] = "agent@yourdomain.com"
        msg["To"] = "you@yourdomain.com"

        with smtplib.SMTP("smtp.yourdomain.com", 587) as server:
            server.starttls()
            server.login("agent@yourdomain.com", "password")
            server.send_message(msg)
Attach it to your agent executor:

from langchain.agents import AgentExecutor

handler = EmailNotificationHandler()
executor = AgentExecutor(agent=agent, tools=tools, callbacks=[handler])
result = executor.invoke({"input": "Research the latest AI funding rounds"})

This works. For about a week. Then you start hitting the problems nobody warns you about in tutorials.

The production problems#

Silent delivery failures#

Your on_agent_finish callback fires. The email dispatch function runs without raising an exception. But the email never arrives. Maybe the SMTP connection timed out and your fire-and-forget code didn't check the response. Maybe the recipient's server rejected the message with a 550 error. Maybe your sending domain has no SPF record and Gmail dropped it into spam without telling anyone.

The callback system has no opinion about whether your email actually landed. It just calls your method and moves on. If the email transport fails silently, the callback still counts as "handled." This is the gap that most tutorials ignore completely: the difference between "the code ran" and "the notification was delivered."

For more on why email transport under AI agents is its own problem, we covered the major failure modes in our guide to LangChain email integration without the OAuth headache.

Duplicate notifications from repeated events#

on_chain_end fires for every chain in your agent's execution, not just the top-level one. If your agent uses three tools, each with their own chain, you get four on_chain_end events (three tool chains plus the outer agent chain). Wire up email notifications to on_chain_end without filtering, and your inbox floods.

Even on_agent_finish isn't always safe from duplicates. Retry logic in your orchestration layer can trigger the finish event multiple times for what the user considers a single task. Track sent notifications by run_id and deduplicate before dispatching: a simple in-memory set works for single-process agents, and Redis or a similar shared cache covers distributed setups.

Blocking the agent's execution#

The synchronous BaseCallbackHandler runs your email code on the same thread as the agent. If your SMTP server takes three seconds to respond, your agent pauses for three seconds mid-run. At scale, this adds up.

from langchain_core.callbacks import AsyncCallbackHandler

class AsyncEmailHandler(AsyncCallbackHandler):
    async def on_agent_finish(self, finish, **kwargs):
        await send_email_async(
            to="you@yourdomain.com",
            subject="Agent finished",
            body=f"Result: {finish.return_values}"
        )

AsyncCallbackHandler fixes the blocking issue. It does not fix delivery reliability or deduplication. Those are separate problems.

LangGraph changes the rules#

If you've migrated to LangGraph (and LangChain is pushing everyone in that direction), callback behavior changes. on_agent_action and on_agent_finish are AgentExecutor-specific events. They simply don't fire in a LangGraph StateGraph. This is why your agent callback might not trigger even when on_tool_start and on_llm_end work correctly.

In LangGraph, you pass callbacks through the config at invocation:

app = graph.compile()
result = app.invoke(
    {"input": "research query"},
    config={"callbacks": [handler]}
)

This is more flexible in some ways, but it means your existing AgentExecutor callback handlers won't work without modification after the migration. Filter by node name or event metadata inside your handler if you need to target specific steps.

A better pattern: decouple notification from transport#

The real issue with putting email logic directly inside a callback handler is coupling. Your agent's execution flow shouldn't depend on whether an SMTP server is responsive. A more reliable approach separates the notification decision from the delivery mechanism.

Instead of sending email from the callback, push a notification event to a webhook endpoint. Let a separate service handle email delivery, retries, and deduplication.

import httpx

class WebhookNotificationHandler(AsyncCallbackHandler):
    async def on_agent_finish(self, finish, **kwargs):
        async with httpx.AsyncClient() as client:
            await client.post(
                "https://your-webhook-endpoint.com/notify",
                json={
                    "event": "agent_finish",
                    "result": str(finish.return_values),
                    "agent_id": str(kwargs.get("run_id", "unknown"))
                }
            )

The webhook endpoint can then format the email, check for duplicates, and hand it off to a transactional email provider that handles SPF, DKIM, bounce processing, and delivery monitoring. This separation means a slow email server never stalls your agent, and a failed delivery can be retried independently.

When the agent itself needs an inbox#

There's a related pattern that gets conflated with callback notifications: giving the agent its own email address so it can send and receive mail as part of its workflow. Not just pinging you when something happens, but actually participating in email conversations.

Think about what agents actually do with their own email. An agent that monitors a shared inbox, extracts action items, and sends follow-ups needs persistent email access. That's not a callback notification problem. That's email infrastructure.

LobsterMail handles this side of things. Your agent provisions its own inbox, sends and receives through it, and you don't configure DNS or SMTP. If you want your LangChain agent to have a working email address it controls, set it up here and paste the instructions to your agent.

For callback-triggered notifications ("tell me when the agent finishes"), the webhook pattern above with a proper email transport layer is what I'd recommend. Use callbacks for the decision, dedicated infrastructure for the delivery.

Choosing the right callback event#

Not all events are equal for notification purposes:

Stick with on_agent_finish for success notifications and on_tool_error for failure alerts. If you need on_chain_end, filter by the chain's run_id or name to avoid duplicate messages.

Testing without sending real emails#

Don't wire up live email during development. Mock the transport:

from datetime import datetime

class TestEmailHandler(BaseCallbackHandler):
    def __init__(self):
        self.sent_emails = []

    def on_agent_finish(self, finish, **kwargs):
        self.sent_emails.append({
            "subject": "Agent finished",
            "body": str(finish.return_values),
            "timestamp": datetime.now()
        })

handler = TestEmailHandler()
executor.invoke({"input": "test query"}, config={"callbacks": [handler]})
assert len(handler.sent_emails) == 1

This lets you verify the callback fires correctly, confirm the right data gets captured, and catch duplicate events before they become duplicate emails hitting a real inbox.

The callback system in LangChain is solid for triggering notifications. Reliable email delivery is a separate concern. Solve them separately, and both work better.