docs(langchain): Docstring improved to show Streaming custom events (#34353)

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(