feat: add post-call integrations framework with slack

bolna/agent_manager/assistant_manager.py

@@ -1,11 +1,13 @@
+import asyncio
 import copy
 import time
 import uuid
 
 from .base_manager import BaseManager
 from .task_manager import TaskManager
 from bolna.helpers.logger_config import configure_logger
-from bolna.models import AGENT_WELCOME_MESSAGE
+from bolna.integrations import PostCallContext, run_post_call_integrations
+from bolna.models import AGENT_WELCOME_MESSAGE, IntegrationConfig
 from bolna.helpers.utils import update_prompt_with_context
 
 logger = configure_logger(__name__)
@@ -40,6 +42,8 @@ def __init__(
         self.output_queue = output_queue
         self.kwargs = kwargs
         self.conversation_history = conversation_history
+        # keep strong refs so fire-and-forget integration tasks are not GC'd
+        self._background_tasks: set = set()
         if kwargs.get("is_web_based_call", False):
             self.kwargs["agent_welcome_message"] = agent_config.get("agent_welcome_message", AGENT_WELCOME_MESSAGE)
         else:
@@ -55,6 +59,7 @@ async def run(self, local=False, run_id=None):
             self.run_id = run_id
 
         input_parameters = None
+        all_task_outputs = []
         for task_id, task in enumerate(self.tasks):
             logger.info(f"Running task {task_id}")
             task_manager = TaskManager(
@@ -81,10 +86,59 @@ async def run(self, local=False, run_id=None):
             )
             task_output = await task_manager.run()
             task_output["run_id"] = self.run_id
+            all_task_outputs.append(task_output)
             yield task_id, copy.deepcopy(task_output)
             self.task_states[task_id] = True
             if task_id == 0:
                 input_parameters = task_output
             if task["task_type"] == "extraction":
                 input_parameters["extraction_details"] = task_output["extracted_data"]
+
+        try:
+            self._fire_post_call_integrations(all_task_outputs)
+        except Exception as e:
+            logger.error(f"failed to schedule post-call integrations: {e}")
+
         logger.info("Done with execution of the agent")
+
+    def _collect_integration_configs(self):
+        seen = set()
+        configs = []
+        for task in self.tasks:
+            tools = task.get("tools_config") or {}
+            for raw in (tools.get("integrations") or []):
+                cfg = raw if isinstance(raw, IntegrationConfig) else IntegrationConfig(**raw)
+                if cfg.provider in seen:
+                    logger.warning(
+                        f"duplicate integration provider '{cfg.provider}' discarded; keeping first config"
+                    )
+                    continue
+                seen.add(cfg.provider)
+                configs.append(cfg)
+        return configs
+
+    def _build_post_call_context(self, all_task_outputs):
+        primary = all_task_outputs[0] if all_task_outputs else {}
+        ctx = PostCallContext(
+            agent_name=self.agent_config.get("agent_name", self.agent_config.get("assistant_name", "")) or "",
+            run_id=self.run_id,
+            call_sid=primary.get("call_sid"),
+            duration_seconds=primary.get("conversation_time"),
+            hangup_reason=str(primary["hangup_detail"]) if primary.get("hangup_detail") else None,
+            recording_url=primary.get("recording_url"),
+        )
+        for output in all_task_outputs:
+            if output.get("task_type") == "summarization" and output.get("summary"):
+                ctx.summary = output["summary"]
+            elif output.get("task_type") == "extraction" and output.get("extracted_data"):
+                ctx.extracted_data = output["extracted_data"]
+        return ctx
+
+    def _fire_post_call_integrations(self, all_task_outputs):
+        configs = self._collect_integration_configs()
+        if not configs:
+            return
+        ctx = self._build_post_call_context(all_task_outputs)
+        task = asyncio.create_task(run_post_call_integrations(configs, ctx))
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)

bolna/enums.py

@@ -173,6 +173,16 @@ def all_values(cls):
         return [e.value for e in cls]
 
 
+class IntegrationProvider(str, Enum):
+    """Enum for post-call integration providers."""
+
+    SLACK = "slack"
+
+    @classmethod
+    def all_values(cls):
+        return [p.value for p in cls]
+
+
 class HangupReason(str, Enum):
     """Enum for hangup_detail values — why the call ended."""
 

bolna/integrations/__init__.py

@@ -0,0 +1,4 @@
+from .base import PostCallContext, PostCallIntegration
+from .runner import run_post_call_integrations
+
+__all__ = ["PostCallContext", "PostCallIntegration", "run_post_call_integrations"]

bolna/integrations/base.py

@@ -0,0 +1,27 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass
+class PostCallContext:
+    agent_name: str
+    run_id: str
+    call_sid: Optional[str] = None
+    duration_seconds: Optional[float] = None
+    hangup_reason: Optional[str] = None
+    summary: Optional[str] = None
+    extracted_data: dict = field(default_factory=dict)
+    recording_url: Optional[str] = None
+
+
+class PostCallIntegration(ABC):
+    @classmethod
+    @abstractmethod
+    def from_config(cls, config) -> "PostCallIntegration":
+        ...
+
+    # may raise; the runner swallows exceptions and logs
+    @abstractmethod
+    async def execute(self, ctx: PostCallContext) -> None:
+        ...

bolna/integrations/runner.py

@@ -0,0 +1,65 @@
+import asyncio
+from typing import List, Optional
+
+import aiohttp
+
+from bolna.helpers.logger_config import configure_logger
+from .base import PostCallContext, PostCallIntegration
+
+logger = configure_logger(__name__)
+
+_RETRY_DELAYS = [1.0, 2.0]
+_REQUEST_TIMEOUT = 10.0
+_INTEGRATION_TIMEOUT = 30.0
+
+
+async def _post_with_retry(url: str, payload: dict, headers: Optional[dict] = None) -> None:
+    last_exc: Optional[Exception] = None
+    timeout = aiohttp.ClientTimeout(total=_REQUEST_TIMEOUT)
+    attempts = len(_RETRY_DELAYS) + 1
+
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        for attempt in range(attempts):
+            try:
+                async with session.post(url, json=payload, headers=headers) as resp:
+                    if resp.status < 400:
+                        return
+                    body = await resp.text()
+                    if resp.status < 500:
+                        logger.error(f"post to {url} failed with {resp.status}: {body[:200]}")
+                        return
+                    last_exc = RuntimeError(f"HTTP {resp.status}: {body[:200]}")
+            except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as e:
+                last_exc = e
+
+            if attempt < attempts - 1:
+                await asyncio.sleep(_RETRY_DELAYS[attempt])
+
+    if last_exc:
+        raise last_exc
+
+
+async def run_post_call_integrations(integration_configs: List, ctx: PostCallContext) -> None:
+    if not integration_configs:
+        return
+
+    from bolna.providers import SUPPORTED_INTEGRATIONS
+
+    for cfg in integration_configs:
+        provider = cfg.provider
+        cls = SUPPORTED_INTEGRATIONS.get(provider)
+        if cls is None:
+            logger.warning(f"unknown integration provider: {provider}")
+            continue
+        try:
+            integration: PostCallIntegration = cls.from_config(cfg)
+        except Exception as e:
+            logger.error(f"could not build {provider} integration: {e}")
+            continue
+
+        try:
+            await asyncio.wait_for(integration.execute(ctx), timeout=_INTEGRATION_TIMEOUT)
+        except asyncio.TimeoutError:
+            logger.error(f"{provider} integration timed out after {_INTEGRATION_TIMEOUT}s")
+        except Exception as e:
+            logger.error(f"{provider} integration failed: {e}")

bolna/integrations/slack.py

@@ -0,0 +1,104 @@
+import os
+from typing import Optional
+
+from bolna.helpers.logger_config import configure_logger
+from .base import PostCallContext, PostCallIntegration
+from .runner import _post_with_retry
+
+logger = configure_logger(__name__)
+
+_MAX_SUMMARY_CHARS = 2800
+_MAX_FIELD_VALUE_CHARS = 250
+
+
+class SlackIntegration(PostCallIntegration):
+    def __init__(self, webhook_url: str):
+        self.webhook_url = webhook_url
+
+    @classmethod
+    def from_config(cls, config) -> "SlackIntegration":
+        webhook_url = config.provider_config.webhook_url or os.getenv("SLACK_WEBHOOK_URL")
+        if not webhook_url:
+            raise ValueError("slack webhook_url not provided in config or SLACK_WEBHOOK_URL env")
+        return cls(webhook_url=webhook_url)
+
+    async def execute(self, ctx: PostCallContext) -> None:
+        payload = {"blocks": _build_slack_blocks(ctx)}
+        await _post_with_retry(self.webhook_url, payload)
+
+
+def _format_duration(seconds: Optional[float]) -> str:
+    if seconds is None or seconds <= 0:
+        return "n/a"
+    minutes, secs = divmod(int(seconds), 60)
+    if minutes:
+        return f"{minutes}m {secs}s"
+    return f"{secs}s"
+
+
+def _truncate(value: str, limit: int) -> str:
+    if len(value) <= limit:
+        return value
+    return value[: limit - 1] + "…"
+
+
+def _build_slack_blocks(ctx: PostCallContext) -> list:
+    blocks = [
+        {
+            "type": "header",
+            "text": {"type": "plain_text", "text": f"call ended — {ctx.agent_name}"},
+        },
+    ]
+
+    context_parts = []
+    if ctx.call_sid:
+        context_parts.append(f"*call_sid:* `{ctx.call_sid}`")
+    context_parts.append(f"*duration:* {_format_duration(ctx.duration_seconds)}")
+    if ctx.hangup_reason:
+        context_parts.append(f"*hangup:* {ctx.hangup_reason}")
+    context_parts.append(f"*run_id:* `{ctx.run_id}`")
+    blocks.append(
+        {
+            "type": "section",
+            "text": {"type": "mrkdwn", "text": "  ·  ".join(context_parts)},
+        }
+    )
+
+    if ctx.summary:
+        blocks.append({"type": "divider"})
+        blocks.append(
+            {
+                "type": "section",
+                "text": {
+                    "type": "mrkdwn",
+                    "text": f"*summary*\n{_truncate(ctx.summary, _MAX_SUMMARY_CHARS)}",
+                },
+            }
+        )
+
+    if ctx.extracted_data:
+        fields = []
+        for key, value in ctx.extracted_data.items():
+            label = str(key).replace("_", " ")
+            text = _truncate(str(value), _MAX_FIELD_VALUE_CHARS) if value not in (None, "") else "_n/a_"
+            fields.append({"type": "mrkdwn", "text": f"*{label}*\n{text}"})
+            if len(fields) == 10:
+                break
+        blocks.append({"type": "divider"})
+        blocks.append({"type": "section", "fields": fields})
+
+    if ctx.recording_url:
+        blocks.append(
+            {
+                "type": "actions",
+                "elements": [
+                    {
+                        "type": "button",
+                        "text": {"type": "plain_text", "text": "open recording"},
+                        "url": ctx.recording_url,
+                    }
+                ],
+            }
+        )
+
+    return blocks

bolna/models.py

@@ -12,6 +12,7 @@
     ExpressionOperator,
     ExpressionLogic,
     EdgeConditionType,
+    IntegrationProvider,
 )
 from .constants import MODEL_REASONING_EFFORT_MAP
 
@@ -464,13 +465,38 @@ class ToolModel(BaseModel):
     tools_params: Dict[str, APIParams]
 
 
+class SlackIntegrationConfig(BaseModel):
+    webhook_url: Optional[str] = None
+
+
+class IntegrationConfig(BaseModel):
+    provider: str
+    provider_config: SlackIntegrationConfig
+
+    @model_validator(mode="before")
+    def preprocess(cls, values):
+        provider = values.get("provider")
+        config = values.get("provider_config", {})
+
+        if provider == IntegrationProvider.SLACK.value:
+            if isinstance(config, dict):
+                values["provider_config"] = SlackIntegrationConfig(**config)
+
+        return values
+
+    @field_validator("provider")
+    def validate_model(cls, value):
+        return validate_attribute(value, IntegrationProvider.all_values(), "integration provider")
+
+
 class ToolsConfig(BaseModel):
     llm_agent: Optional[Union[LlmAgent, SimpleLlmAgent]] = None
     synthesizer: Optional[Synthesizer] = None
     transcriber: Optional[Transcriber] = None
     input: Optional[IOModel] = None
     output: Optional[IOModel] = None
     api_tools: Optional[ToolModel] = None
+    integrations: Optional[List[IntegrationConfig]] = None
     switch_tool_description: Optional[str] = None
     switch_handoff_messages: Optional[Dict[str, str]] = None
     agent_names: Optional[Dict[str, str]] = None

bolna/providers.py

@@ -38,7 +38,8 @@
     SipTrunkOutputHandler,
 )
 from .llms import OpenAiLLM, LiteLLM, AzureLLM, GeminiLLM
-from .enums import TelephonyProvider, SynthesizerProvider, TranscriberProvider, LLMProvider
+from .integrations.slack import SlackIntegration
+from .enums import TelephonyProvider, SynthesizerProvider, TranscriberProvider, LLMProvider, IntegrationProvider
 
 SUPPORTED_SYNTHESIZER_MODELS = {
     SynthesizerProvider.POLLY.value: PollySynthesizer,
@@ -118,3 +119,6 @@
     TelephonyProvider.VOBIZ.value: VobizOutputHandler,
     TelephonyProvider.SIP_TRUNK.value: SipTrunkOutputHandler,
 }
+SUPPORTED_INTEGRATIONS = {
+    IntegrationProvider.SLACK.value: SlackIntegration,
+}