From 4804bd6ec23390d0fc1d22b7f42bcf877a633b8e Mon Sep 17 00:00:00 2001
From: Mohammad Mohtashim <45242107+keenborder786@users.noreply.github.com>
Date: Sat, 20 Dec 2025 00:15:10 +0500
Subject: [PATCH] docs(langchain): Docstring improved to show `Streaming custom
 events` (#34353)

---
 .../langchain/agents/middleware/types.py      | 91 +++++++++++++++++++
 1 file changed, 91 insertions(+)

diff --git a/libs/langchain_v1/langchain/agents/middleware/types.py b/libs/langchain_v1/langchain/agents/middleware/types.py
index ebb1ad4ee76..eefdb7c4571 100644
--- a/libs/langchain_v1/langchain/agents/middleware/types.py
+++ b/libs/langchain_v1/langchain/agents/middleware/types.py
@@ -883,6 +883,23 @@ def before_model(
             def custom_before_model(state: MyCustomState, runtime: Runtime) -> dict[str, Any]:
                 return {"custom_field": "updated_value"}
             ```
+
+        !!! example "Streaming custom events before model call"
+
+            Use `runtime.stream_writer` to emit custom events before each model invocation.
+            Events are received when streaming with `stream_mode="custom"`.
+
+            ```python
+            @before_model
+            async def notify_model_call(state: AgentState, runtime: Runtime) -> None:
+                '''Notify user before model is called.'''
+                runtime.stream_writer(
+                    {
+                        "type": "status",
+                        "message": "Thinking...",
+                    }
+                )
+            ```
     """
 
     def decorator(
@@ -1024,6 +1041,25 @@ def after_model(
             def custom_after_model(state: MyCustomState, runtime: Runtime) -> dict[str, Any]:
                 return {"custom_field": "updated_after_model"}
             ```
+
+        !!! example "Streaming custom events after model call"
+
+            Use `runtime.stream_writer` to emit custom events after model responds.
+            Events are received when streaming with `stream_mode="custom"`.
+
+            ```python
+            @after_model
+            async def notify_model_response(state: AgentState, runtime: Runtime) -> None:
+                '''Notify user after model has responded.'''
+                last_message = state["messages"][-1]
+                has_tool_calls = hasattr(last_message, "tool_calls") and last_message.tool_calls
+                runtime.stream_writer(
+                    {
+                        "type": "status",
+                        "message": "Using tools..." if has_tool_calls else "Response ready!",
+                    }
+                )
+            ```
     """
 
     def decorator(
@@ -1175,6 +1211,43 @@ def before_agent(
             def custom_before_agent(state: MyCustomState, runtime: Runtime) -> dict[str, Any]:
                 return {"custom_field": "initialized_value"}
             ```
+
+        !!! example "Streaming custom events"
+
+            Use `runtime.stream_writer` to emit custom events during agent execution.
+            Events are received when streaming with `stream_mode="custom"`.
+
+            ```python
+            from langchain.agents import create_agent
+            from langchain.agents.middleware import before_agent, AgentState
+            from langchain.messages import HumanMessage
+            from langgraph.runtime import Runtime
+
+
+            @before_agent
+            async def notify_start(state: AgentState, runtime: Runtime) -> None:
+                '''Notify user that agent is starting.'''
+                runtime.stream_writer(
+                    {"type": "status", "message": "Initializing agent session..."}
+                )
+                # Perform prerequisite tasks here
+                runtime.stream_writer({"type": "status", "message": "Agent ready!"})
+
+
+            agent = create_agent(
+                model="openai:gpt-5.2",
+                tools=[...],
+                middleware=[notify_start],
+            )
+
+            # Consume with stream_mode="custom" to receive events
+            async for mode, event in agent.astream(
+                {"messages": [HumanMessage("Hello")]},
+                stream_mode=["updates", "custom"],
+            ):
+                if mode == "custom":
+                    print(f"Status: {event}")
+            ```
     """
 
     def decorator(
@@ -1318,6 +1391,24 @@ def after_agent(
             def custom_after_agent(state: MyCustomState, runtime: Runtime) -> dict[str, Any]:
                 return {"custom_field": "finalized_value"}
             ```
+
+        !!! example "Streaming custom events on completion"
+
+            Use `runtime.stream_writer` to emit custom events when agent completes.
+            Events are received when streaming with `stream_mode="custom"`.
+
+            ```python
+            @after_agent
+            async def notify_completion(state: AgentState, runtime: Runtime) -> None:
+                '''Notify user that agent has completed.'''
+                runtime.stream_writer(
+                    {
+                        "type": "status",
+                        "message": "Agent execution complete!",
+                        "total_messages": len(state["messages"]),
+                    }
+                )
+            ```
     """
 
     def decorator(