How to build a webhook-driven agent router service in LangChain

Webhook endpoints are often the first place an external system hands work to an agent application. A small router service can turn billing, monitoring, or audit events into different LangChain agent actions while keeping the HTTP contract separate from model prompts.

FastAPI receives and validates the JSON request before LangChain sees it. The service combines a Pydantic request model, event-type routing, LangChain tools, and create_agent() so each webhook branch returns a named route, tool output, and final agent reply.

The local smoke test uses FakeMessagesListChatModel to avoid provider keys and keep the routing proof repeatable. Replace that fake model with a provider-backed chat model after the payload schema, branch mapping, and tool side effects are safe for the source systems that call the webhook.

1. Open an activated Python project environment.

   Related: How to create a virtual environment for LangChain

2. Install LangChain, FastAPI, and Uvicorn.

   $ python3 -m pip install --upgrade langchain fastapi "uvicorn[standard]"

   LangChain requires Python 3.10 or newer. Provider integrations are separate packages when the router uses a live model provider.
   Related: How to install LangChain with pip

3. Create app.py with the webhook schema, tools, router, and FastAPI endpoint.

   app.py

   from typing import Any, Literal
    
   from fastapi import FastAPI
   from langchain.agents import create_agent
   from langchain.tools import tool
   from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
   from langchain_core.messages import AIMessage, ToolMessage
   from pydantic import BaseModel, Field
    
    
   class WebhookEvent(BaseModel):
       source: str = Field(min_length=2)
       event_type: Literal["ticket.created", "incident.alert", "audit.recorded"]
       payload: dict[str, Any] = Field(default_factory=dict)
    
    
   class RouterResult(BaseModel):
       route: str
       tool: str
       tool_output: str
       agent_reply: str
    
    
   @tool
   def summarize_ticket(title: str, priority: str = "normal") -> str:
       """Create a support ticket routing summary."""
       return f"support ticket routed: {title} ({priority})"
    
    
   @tool
   def escalate_incident(service: str, severity: str) -> str:
       """Create an incident escalation note."""
       return f"incident escalation queued for {service} at {severity} severity"
    
    
   @tool
   def record_audit_event(action: str, actor: str) -> str:
       """Record a security audit event."""
       return f"audit event recorded: {actor} {action}"
    
    
   TOOLS = [summarize_ticket, escalate_incident, record_audit_event]
    
    
   class RoutingDemoModel(FakeMessagesListChatModel):
       def bind_tools(self, tools, **kwargs):
           return self
    
    
   def route_for(event: WebhookEvent) -> dict[str, Any]:
       if event.event_type == "ticket.created":
           return {
               "name": "support_triage",
               "tool": "summarize_ticket",
               "args": {
                   "title": str(event.payload.get("title", "Untitled ticket")),
                   "priority": str(event.payload.get("priority", "normal")),
               },
           }
       if event.event_type == "incident.alert":
           return {
               "name": "incident_escalation",
               "tool": "escalate_incident",
               "args": {
                   "service": str(event.payload.get("service", "unknown-service")),
                   "severity": str(event.payload.get("severity", "warning")),
               },
           }
       return {
           "name": "audit_log",
           "tool": "record_audit_event",
           "args": {
               "action": str(event.payload.get("action", "unknown-action")),
               "actor": str(event.payload.get("actor", event.source)),
           },
       }
    
    
   def invoke_routed_agent(event: WebhookEvent) -> RouterResult:
       route = route_for(event)
       model = RoutingDemoModel(
           responses=[
               AIMessage(
                   content="",
                   tool_calls=[
                       {
                           "id": f"call_{route['tool']}",
                           "name": route["tool"],
                           "args": route["args"],
                       }
                   ],
               ),
               AIMessage(
                   content=(
                       f"Agent route {route['name']} completed for "
                       f"{event.event_type}."
                   )
               ),
           ]
       )
       agent = create_agent(
           model=model,
           tools=TOOLS,
           system_prompt="Route webhook events to the matching operational tool.",
       )
       result = agent.invoke(
           {
               "messages": [
                   {
                       "role": "user",
                       "content": (
                           f"Handle {event.event_type} from {event.source} "
                           f"with payload {event.payload}."
                       ),
                   }
               ]
           }
       )
       tool_output = next(
           message.content
           for message in result["messages"]
           if isinstance(message, ToolMessage)
       )
       return RouterResult(
           route=route["name"],
           tool=route["tool"],
           tool_output=tool_output,
           agent_reply=result["messages"][-1].content,
       )
    
    
   app = FastAPI(title="LangChain Webhook Agent Router")
    
    
   @app.get("/health")
   def health() -> dict[str, str]:
       return {"status": "ok"}
    
    
   @app.post("/webhooks/events", response_model=RouterResult)
   def handle_webhook(event: WebhookEvent) -> RouterResult:
       return invoke_routed_agent(event)

   route_for() keeps event classification deterministic. The fake model emits the selected tool call for local testing; a provider model can replace RoutingDemoModel after the route and tool contracts are proven.

4. Check that Python can load the service file.

   $ python3 -m py_compile app.py

   No output means Python accepted the file syntax.

5. Start the webhook router service with Uvicorn.

   $ python3 -m uvicorn app:app --host 127.0.0.1 --port 8000
   INFO:     Started server process [214]
   INFO:     Waiting for application startup.
   INFO:     Application startup complete.
   INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

   127.0.0.1 keeps the development service local to the machine. Bind to a wider interface only behind the ingress controls used for the webhook source.

6. Send a support-ticket webhook event.

   $ curl --silent --show-error --request POST http://127.0.0.1:8000/webhooks/events \
     --header 'Content-Type: application/json' \
     --data '{"source":"stripe","event_type":"ticket.created","payload":{"title":"Billing failure","priority":"high"}}'
   {"route":"support_triage","tool":"summarize_ticket","tool_output":"support ticket routed: Billing failure (high)","agent_reply":"Agent route support_triage completed for ticket.created."}

7. Send an incident webhook event and confirm the router selects a different agent branch.

   $ curl --silent --show-error --request POST http://127.0.0.1:8000/webhooks/events \
     --header 'Content-Type: application/json' \
     --data '{"source":"monitoring","event_type":"incident.alert","payload":{"service":"checkout-api","severity":"critical"}}'
   {"route":"incident_escalation","tool":"escalate_incident","tool_output":"incident escalation queued for checkout-api at critical severity","agent_reply":"Agent route incident_escalation completed for incident.alert."}

   The event_type field controls the selected route. Unknown event types fail request validation before the agent runs.

8. Stop the local service after the webhook smoke test.

   Press Ctrl-C in the terminal running Uvicorn.