# How to Implement Webhook Callbacks for Async LLM Task Completion Notifications in LMForge > Learn to implement webhook callbacks for async LLM task completion. LMForge extends schemas to notify external endpoints upon task stream completion. Streamline your LLMOps. - Repository: [Haohao/lmforge-end-to-end-llmops-platform-for-multi-model-agents](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents) - Tags: how-to-guide - Published: 2026-03-03 --- **You can implement webhook callbacks for async LLM task completion notifications by extending the request schema with an optional `webhook_url` field, capturing it in the `WebAppService` streaming generator, and invoking a `_notify_webhook` helper method that POSTs a JSON payload to the external endpoint after the task stream ends.** The LMForge platform executes LLM-driven tasks asynchronously using **Celery** workers and streams intermediate results to the frontend via **Server-Sent Events (SSE)**. To integrate with external services that require notification when a task completes or fails, you can implement webhook callbacks for async LLM task completion notifications that fire server-side without disrupting the real-time user experience. ## Architecture Overview The platform's task lifecycle provides three distinct integration points for webhook delivery. According to the LMForge source code, the execution flow runs through [`api/internal/task/app_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/task/app_task.py) where Celery tasks like `web_app_chat` are declared with `@shared_task`, then delegates to [`api/internal/service/web_app_service.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/service/web_app_service.py) which yields `AgentThought` objects until a terminal event occurs. The `QueueEvent` enum defined in [`api/internal/core/agent/entities/queue_entity.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/core/agent/entities/queue_entity.py) identifies final states including `STOP`, `ERROR`, `AGENT_END`, and `TIMEOUT`. By hooking into the completion of the generator loop in the service layer, you can reliably trigger a single POST request to a caller-supplied URL. ## Step-by-Step Implementation ### Update the Request Schema First, extend `WebAppChatReq` in [`api/internal/schema/web_app_schema.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/schema/web_app_schema.py) to accept an optional webhook URL. The platform uses **WTForms** for validation, so add a `StringField` with URL validation: ```python # api/internal/schema/web_app_schema.py from wtforms import StringField from wtforms.validators import Optional, URL class WebAppChatReq(FlaskForm): ... webhook_url = StringField( "webhook_url", default="", validators=[Optional(), URL(message="Invalid webhook URL")] ) ``` ### Capture the Webhook URL in the Service Layer Modify `WebAppService.web_app_chat` in [`api/internal/service/web_app_service.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/service/web_app_service.py) to extract the URL and store it in the method context. This method already yields SSE rows via a generator; you will invoke the notifier after the `for` loop exhausts the `agent.stream()` iterator: ```python # api/internal/service/web_app_service.py def web_app_chat(self, token: str, req: WebAppChatReq, account: Account) -> Generator: webhook_url = req.webhook_url.data or None task_id = None for agent_thought in agent.stream(...): task_id = str(agent_thought.task_id) data = { "event": agent_thought.event.value, "thought": agent_thought.thought, } yield f"event: {agent_thought.event.value}\ndata:{json.dumps(data)}\n\n" # Trigger webhook after stream completion if webhook_url and task_id: self._notify_webhook( webhook_url, task_id=task_id, status="completed", payload={"conversation_id": str(account.id)} ) ``` ### Build the Webhook Notifier Implement a private `_notify_webhook` method inside `WebAppService` (or refactor into a shared `WebhookService` mixin if multiple services require this behavior). This method uses Python's `requests` library for a fire-and-forget HTTP call: ```python # api/internal/service/web_app_service.py import requests from .base_service import BaseService class WebAppService(BaseService): ... def _notify_webhook( self, url: str, task_id: str, status: str, payload: dict | None = None ): body = { "task_id": task_id, "status": status, "detail": payload or {}, } try: # Fire-and-forget with short timeout to avoid blocking requests.post(url, json=body, timeout=3) except Exception as exc: # Log but do not raise to preserve user-facing flow self.logger.warning(f"Webhook delivery failed for {url}: {exc}") ``` ### Detect Task Completion Events The streaming loop terminates when `AgentThought.event` matches a terminal value from the `QueueEvent` enum. The final `agent_thought` object remains accessible after the loop exits, allowing you to inspect its state to determine whether the task succeeded, was stopped by the user, or encountered an error: ```python from api.internal.core.agent.entities.queue_entity import QueueEvent # Inside web_app_chat method, after the for loop final_status = "completed" if agent_thought.event == QueueEvent.ERROR.value: final_status = "failed" elif agent_thought.event == QueueEvent.STOP.value: final_status = "stopped" if webhook_url: self._notify_webhook( webhook_url, task_id=str(agent_thought.task_id), status=final_status, payload={"final_event": agent_thought.event.value} ) ``` ## Frontend Integration Example External callers include the `webhook_url` parameter when initializing a chat session. The existing `handleWebAppChat` utility in the frontend accepts this field and passes it through to the Flask backend: ```typescript // src/views/web-apps/IndexView.vue const webhookUrl = 'https://example.com/llm-callback'; await handleWebAppChat( token, { conversation_id: currentConversationId, query: userPrompt, webhook_url: webhookUrl, // Added field }, (event) => { // Existing SSE handling for real-time updates } ); ``` The client continues to receive intermediate `AgentThought` events via SSE while the server-side webhook fires only once after the stream closes. ## Webhook Payload Structure The JSON payload posted to the webhook URL contains the task identifier, terminal status, and optional metadata: ```json { "task_id": "03042ead-23d6-42c6-a5c9-caac416bb80c", "status": "completed", "detail": { "conversation_id": "76cf52b8-36b7-4fd5-af8b-65547991f84a", "final_event": "AGENT_END", "answer": "The final answer generated by the LLM …" } } ``` External services should implement idempotent handlers that act on `task_id` and `status` without requiring additional state from the LMForge platform. ## Summary - **Extend the schema** by adding `webhook_url` to `WebAppChatReq` in [`api/internal/schema/web_app_schema.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/schema/web_app_schema.py) with URL validation. - **Capture the URL** in `WebAppService.web_app_chat` and store it for use after the generator loop completes. - **Implement `_notify_webhook`** using `requests.post` with a short timeout to avoid blocking the SSE stream. - **Trigger on terminal events** identified by the `QueueEvent` enum (`AGENT_END`, `STOP`, `ERROR`, `TIMEOUT`) to ensure exactly one callback per task. - **Reuse the pattern** across other async services (assistant-agent, app debug) by extracting the notifier into a shared mixin or `WebhookService`. ## Frequently Asked Questions ### What happens if the webhook endpoint is unreachable? The implementation uses a try/except block that logs the failure via `self.logger.warning` but does not raise an exception. This ensures that temporary network issues or downstream server errors do not disrupt the user-facing chat response or cause the Celery task to retry unnecessarily. ### Can I use webhook callbacks for other LLM services beyond WebApp chats? Yes. The same pattern applies to `assistant_agent_chat` and other Celery tasks defined in [`api/internal/task/app_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/task/app_task.py). Simply add the `webhook_url` field to the corresponding request schema (e.g., `AssistantAgentChatReq` in [`api/internal/schema/assistant_agent_schema.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/schema/assistant_agent_schema.py)) and invoke `_notify_webhook` after the streaming loop in the respective service class. ### Which task events trigger the webhook notification? The webhook fires after the generator exhausts, which occurs when the `AgentThought.event` equals `QueueEvent.AGENT_END`, `STOP`, `ERROR`, or `TIMEOUT`. You can customize the logic to send different statuses based on the specific terminal event, enabling downstream systems to distinguish between successful completions, user cancellations, and runtime errors. ### Does the webhook block the SSE stream from reaching the frontend? No. The webhook POST executes after the `yield` loop finishes and the SSE connection closes. The `requests.post` call uses a 3-second timeout and runs synchronously within the service method, but because it occurs after all SSE data has been flushed to the client, it does not introduce latency into the real-time streaming experience.