How to Implement Webhook Callbacks for Async LLM Task Completion Notifications in LMForge

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 where Celery tasks like web_app_chat are declared with @shared_task, then delegates to 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 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 to accept an optional webhook URL. The platform uses WTForms for validation, so add a StringField with URL validation:

# 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 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:

# 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:

# 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:

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:

// 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:

{
  "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 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. Simply add the webhook_url field to the corresponding request schema (e.g., AssistantAgentChatReq in 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}