Merge remote-tracking branch 'origin/main' into feat/text-chat

This commit is contained in:
Abhishek Kumar 2026-05-21 07:47:07 +05:30
commit 129a6d700c
160 changed files with 9287 additions and 3935 deletions

View file

@ -129,22 +129,6 @@ async def get_user(
return user_model
async def get_user_optional(
authorization: Annotated[str | None, Header()] = None,
x_api_key: Annotated[str | None, Header(alias="X-API-Key")] = None,
) -> UserModel | None:
"""
Same as get_user but returns None instead of raising 401 if unauthorized.
Useful for endpoints that need to work both with and without auth.
"""
try:
return await get_user(authorization, x_api_key)
except HTTPException as e:
if e.status_code == 401:
return None
raise
async def _handle_oss_auth(authorization: str | None) -> UserModel:
"""
Handle authentication for OSS deployment mode.

View file

@ -183,9 +183,7 @@ class CampaignSourceSyncService(ABC):
async def get_source_credentials(
self, organization_id: int, source_type: str
) -> Dict[str, Any]:
"""Gets OAuth tokens or API credentials via Nango"""
# This would be implemented to work with Nango service
# For now, returning placeholder
"""Gets source credentials when a sync service requires them."""
logger.info(
f"Getting credentials for org {organization_id}, source {source_type}"
)

View file

@ -1,15 +1,12 @@
from api.services.campaign.source_sync import CampaignSourceSyncService
from api.services.campaign.sources.csv import CSVSyncService
from api.services.campaign.sources.google_sheets import GoogleSheetsSyncService
def get_sync_service(source_type: str) -> CampaignSourceSyncService:
"""Returns appropriate sync service based on source type"""
services = {
"google-sheet": GoogleSheetsSyncService,
"csv": CSVSyncService,
# Add more as needed: "hubspot": HubSpotSyncService,
}
service_class = services.get(source_type)

View file

@ -1,5 +1,3 @@
"""Campaign source sync services"""
from .google_sheets import GoogleSheetsSyncService
__all__ = ["GoogleSheetsSyncService"]
__all__: list[str] = []

View file

@ -1,224 +0,0 @@
import re
from typing import Any, Dict, List, Optional
import httpx
from loguru import logger
from api.db import db_client
from api.services.campaign.source_sync import (
CampaignSourceSyncService,
ValidationError,
ValidationResult,
)
from api.services.integrations.nango import NangoService
class GoogleSheetsSyncService(CampaignSourceSyncService):
"""Implementation for Google Sheets synchronization"""
def __init__(self):
self.nango_service = NangoService()
self.sheets_api_base = "https://sheets.googleapis.com/v4/spreadsheets"
async def _get_access_token(self, organization_id: int) -> str:
"""Get OAuth access token for Google Sheets via Nango."""
integrations = await db_client.get_integrations_by_organization_id(
organization_id
)
integration = None
for intg in integrations:
if intg.provider == "google-sheet" and intg.is_active:
integration = intg
break
if not integration:
raise ValueError("Google Sheets integration not found or inactive")
token_data = await self.nango_service.get_access_token(
connection_id=integration.integration_id, provider_config_key="google-sheet"
)
return token_data["credentials"]["access_token"]
async def _fetch_all_sheet_data(
self, sheet_url: str, organization_id: int
) -> List[List[str]]:
"""Fetch all data from a Google Sheet. Returns all rows including header."""
access_token = await self._get_access_token(organization_id)
sheet_id = self._extract_sheet_id(sheet_url)
metadata = await self._get_sheet_metadata(sheet_id, access_token)
if not metadata.get("sheets"):
raise ValueError("No sheets found in the spreadsheet")
sheet_name = metadata["sheets"][0]["properties"]["title"]
return await self._fetch_sheet_data(sheet_id, f"{sheet_name}!A:Z", access_token)
async def validate_source(
self, source_id: str, organization_id: Optional[int] = None
) -> ValidationResult:
"""Validate a Google Sheet source for campaign creation."""
if organization_id is None:
return ValidationResult(
is_valid=False,
error=ValidationError(
message="Organization ID is required for Google Sheets validation"
),
)
# Validate URL format first
pattern = r"/spreadsheets/d/([a-zA-Z0-9-_]+)"
if not re.search(pattern, source_id):
return ValidationResult(
is_valid=False,
error=ValidationError(
message=f"Invalid Google Sheets URL: {source_id}"
),
)
try:
rows = await self._fetch_all_sheet_data(source_id, organization_id)
except ValueError as e:
return ValidationResult(
is_valid=False,
error=ValidationError(message=str(e)),
)
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error fetching Google Sheet: {e.response.status_code}")
return ValidationResult(
is_valid=False,
error=ValidationError(
message=f"Failed to fetch Google Sheet data: {e.response.status_code}"
),
)
except Exception as e:
logger.error(f"Error fetching Google Sheet: {e}")
return ValidationResult(
is_valid=False,
error=ValidationError(message="Failed to fetch Google Sheet data"),
)
if not rows or len(rows) < 2:
return ValidationResult(
is_valid=False,
error=ValidationError(
message="Google Sheet must have a header row and at least one data row"
),
)
headers = rows[0]
data_rows = rows[1:]
return self.validate_source_data(headers, data_rows)
async def sync_source_data(self, campaign_id: int) -> int:
"""
Fetches data from Google Sheets and creates queued_runs
"""
# Get campaign
campaign = await db_client.get_campaign_by_id(campaign_id)
if not campaign:
raise ValueError(f"Campaign {campaign_id} not found")
rows = await self._fetch_all_sheet_data(
campaign.source_id, campaign.organization_id
)
if not rows or len(rows) < 2:
logger.warning(f"No data found in sheet for campaign {campaign_id}")
return 0
headers = self.normalize_headers(rows[0])
data_rows = rows[1:]
sheet_id = self._extract_sheet_id(campaign.source_id)
queued_runs = []
for idx, row_values in enumerate(data_rows, 1):
# Pad row to match headers length
padded_row = row_values + [""] * (len(headers) - len(row_values))
# Create context variables dict
context_vars = dict(zip(headers, padded_row))
# Skip if no phone number
if not context_vars.get("phone_number"):
logger.debug(f"Skipping row {idx}: no phone_number")
continue
# Generate unique source UUID
source_uuid = f"sheet_{sheet_id}_row_{idx}"
queued_runs.append(
{
"campaign_id": campaign_id,
"source_uuid": source_uuid,
"context_variables": context_vars,
"state": "queued",
}
)
# Bulk insert
if queued_runs:
await db_client.bulk_create_queued_runs(queued_runs)
logger.info(
f"Created {len(queued_runs)} queued runs for campaign {campaign_id}"
)
# Update campaign total_rows
await db_client.update_campaign(
campaign_id=campaign_id,
total_rows=len(queued_runs),
source_sync_status="completed",
)
return len(queued_runs)
async def _fetch_sheet_data(
self, sheet_id: str, range: str, access_token: str
) -> List[List[str]]:
"""Fetch data from Google Sheets API"""
url = f"{self.sheets_api_base}/{sheet_id}/values/{range}"
headers = {"Authorization": f"Bearer {access_token}"}
async with httpx.AsyncClient() as client:
response = await client.get(url, headers=headers)
response.raise_for_status()
data = response.json()
return data.get("values", [])
async def _get_sheet_metadata(
self, sheet_id: str, access_token: str
) -> Dict[str, Any]:
"""Get sheet metadata including sheet names"""
url = f"{self.sheets_api_base}/{sheet_id}"
headers = {"Authorization": f"Bearer {access_token}"}
logger.debug(f"Fetching sheet metadata from URL: {url}")
logger.debug(f"Using sheet_id: {sheet_id}")
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error {e.response.status_code} for URL: {url}")
logger.error(f"Response body: {e.response.text}")
raise
except Exception as e:
logger.error(f"Error fetching sheet metadata: {e}")
raise
def _extract_sheet_id(self, sheet_url: str) -> str:
"""
Extract sheet ID from various Google Sheets URL formats:
- https://docs.google.com/spreadsheets/d/{id}/edit
- https://docs.google.com/spreadsheets/d/{id}/edit#gid=0
"""
pattern = r"/spreadsheets/d/([a-zA-Z0-9-_]+)"
match = re.search(pattern, sheet_url)
if match:
return match.group(1)
raise ValueError(f"Invalid Google Sheets URL: {sheet_url}")

View file

@ -13,6 +13,7 @@ from typing import Any, Dict, Optional
from api.schemas.user_configuration import UserConfiguration
from api.services.configuration.registry import ServiceConfig
from api.services.integrations import get_node_secret_fields
VISIBLE_CHARS = 4 # number of trailing characters to reveal
MASK_CHAR = "*"
@ -129,14 +130,22 @@ def mask_user_config(config: UserConfiguration) -> Dict[str, Any]:
# ---------------------------------------------------------------------------
# Workflow definition helpers mask / merge QA-node API keys
# Workflow definition helpers mask / merge node API keys
# ---------------------------------------------------------------------------
_QA_API_KEY_FIELD = "qa_api_key"
_NODE_SECRET_FIELDS: dict[str, tuple[str, ...]] = {
"qa": ("qa_api_key",),
}
def _secret_fields_for_node_type(node_type: str | None) -> tuple[str, ...]:
if not node_type:
return ()
return _NODE_SECRET_FIELDS.get(node_type, ()) or get_node_secret_fields(node_type)
def mask_workflow_definition(workflow_definition: Optional[Dict]) -> Optional[Dict]:
"""Return a *shallow copy* of *workflow_definition* with QA-node API keys masked."""
"""Return a copy of *workflow_definition* with node secret fields masked."""
if not workflow_definition:
return workflow_definition
@ -144,47 +153,46 @@ def mask_workflow_definition(workflow_definition: Optional[Dict]) -> Optional[Di
masked = copy.deepcopy(workflow_definition)
for node in masked.get("nodes", []):
if node.get("type") != "qa":
secret_fields = _secret_fields_for_node_type(node.get("type"))
if not secret_fields:
continue
data = node.get("data", {})
raw_key = data.get(_QA_API_KEY_FIELD)
if raw_key:
data[_QA_API_KEY_FIELD] = mask_key(raw_key)
for field in secret_fields:
raw_key = data.get(field)
if raw_key:
data[field] = mask_key(raw_key)
return masked
def merge_workflow_api_keys(
incoming_definition: Optional[Dict], existing_definition: Optional[Dict]
) -> Optional[Dict]:
"""Preserve real QA-node API keys when the incoming value is a masked placeholder.
For each QA node in *incoming_definition*, if its ``qa_api_key`` equals
the masked form of the corresponding node in *existing_definition*, the
real key is restored so it is never lost.
"""
"""Preserve real node secret fields when the incoming value is masked."""
if not incoming_definition or not existing_definition:
return incoming_definition
# Build lookup: node-id → data for existing QA nodes
existing_qa: Dict[str, Dict] = {}
existing_nodes: Dict[str, Dict] = {}
for node in existing_definition.get("nodes", []):
if node.get("type") == "qa":
existing_qa[node["id"]] = node.get("data", {})
if _secret_fields_for_node_type(node.get("type")):
existing_nodes[node["id"]] = node.get("data", {})
for node in incoming_definition.get("nodes", []):
if node.get("type") != "qa":
secret_fields = _secret_fields_for_node_type(node.get("type"))
if not secret_fields:
continue
data = node.get("data", {})
incoming_key = data.get(_QA_API_KEY_FIELD)
if not incoming_key:
continue
old_data = existing_qa.get(node["id"])
old_data = existing_nodes.get(node["id"])
if not old_data:
continue
old_key = old_data.get(_QA_API_KEY_FIELD, "")
if old_key and is_mask_of(incoming_key, old_key):
data[_QA_API_KEY_FIELD] = old_key
for field in secret_fields:
incoming_key = data.get(field)
if not incoming_key:
continue
old_key = old_data.get(field, "")
if old_key and is_mask_of(incoming_key, old_key):
data[field] = old_key
return incoming_definition

View file

@ -975,29 +975,67 @@ class SarvamSTTConfiguration(BaseSTTConfiguration):
# Speechmatics STT Service
SPEECHMATICS_STT_LANGUAGES = [
"en",
"es",
"fr",
"de",
"it",
"pt",
"ar",
"ar_en",
"ba",
"eu",
"be",
"bn",
"bg",
"yue",
"ca",
"hr",
"cs",
"da",
"nl",
"en",
"eo",
"et",
"fi",
"fr",
"gl",
"de",
"el",
"he",
"hi",
"hu",
"id",
"ia",
"ga",
"it",
"ja",
"ko",
"zh",
"ru",
"ar",
"hi",
"pl",
"tr",
"vi",
"th",
"id",
"lv",
"lt",
"ms",
"sv",
"da",
"en_ms",
"mt",
"cmn",
"cmn_en",
"cmn_en_ms_ta",
"mr",
"mn",
"no",
"fi",
"fa",
"pl",
"pt",
"ro",
"ru",
"sk",
"sl",
"es",
"sw",
"sv",
"tl",
"ta",
"en_ta",
"th",
"tr",
"uk",
"ur",
"ug",
"vi",
"cy",
]

View file

@ -0,0 +1,239 @@
# Integrations - Plugin Contract
`api/services/integrations/` is the extension seam for third-party integrations.
New integrations should be self-contained here. Do not bleed integration-specific
logic into `workflow/dto.py`, `workflow/node_specs/`, `run_pipeline.py`,
`event_handlers.py`, or `run_integrations.py` unless you are changing the generic
framework itself.
## Golden Path
Create a package:
```text
api/services/integrations/<name>/
├── __init__.py
├── node.py
├── runtime.py # optional
├── completion.py # optional
├── routes.py # optional
└── client.py # optional
```
The package self-registers on import via `register_package(...)`. Discovery is
automatic: `api/services/integrations/loader.py` imports every submodule under
`api.services.integrations` except the reserved internal names `base`, `loader`,
and `registry`.
## Registration Pattern
`__init__.py` should register one `IntegrationPackageSpec`, following the
existing integration packages in this directory.
Use:
```python
PACKAGE = register_package(
IntegrationPackageSpec(
name="<package_name>",
nodes=(NODE,),
create_runtime_sessions=create_runtime_sessions, # optional
run_completion=run_completion, # optional
routers=(router,), # optional
)
)
```
The package name is the registry key. The node `type_name` is the workflow node
type string and must stay stable once exposed.
## Node Model + Spec
For integration nodes, the Pydantic model is the source of truth. The serialized
`NodeSpec` is derived from it.
Refer to an existing integration node for the overall structure:
- Define one Pydantic model per node, inheriting
`api/services/workflow/node_data.py:BaseNodeData`.
- Annotate it with `@node_spec(...)`.
- Define fields with `spec_field(...)`.
- Generate the external spec with `SPEC = build_spec(ModelClass)`.
- Register the node with `IntegrationNodeRegistration(...)`.
Important rules:
- Put runtime validation in the model, not in the generated spec.
Example: conditional requiredness belongs in `@model_validator(mode="after")`.
- Keep `@node_spec(name=...)` and `IntegrationNodeRegistration.type_name`
identical. They are the same workflow node type string.
- Put wire constraints in the field itself where possible.
Example: `gt=0`, `min_length=1`, `pattern=...`.
- Put UI/export-only differences in `field_overrides`.
Use this for `display_name`, `description`, `required`, `spec_default`,
`display_options`, or property ordering.
- Use `spec_exclude=True` for internal fields that must exist in persisted data
but must not show up in `/api/v1/node-types`.
- Set `property_order=(...)` in `@node_spec(...)` when the editor field order
must remain stable.
Typical workflow graph constraints for configuration-only integration nodes:
```python
GraphConstraints(min_incoming=0, max_incoming=0, min_outgoing=0, max_outgoing=0)
```
These constraints control how the node can be connected in the workflow graph.
Use them for configuration nodes that are not conversational graph steps.
## Secret Fields
If the node stores secrets, register them in
`IntegrationNodeRegistration.sensitive_fields`.
That is enough for generic masking / masked round-trip preservation via
`api/services/configuration/masking.py`. Do not add new integration-specific
masking branches unless you are changing the shared masking framework.
## No Central DTO Edits
Do not add integration node classes to `api/services/workflow/dto.py`.
Integration nodes are resolved dynamically through:
- `get_node_data_model()` in `workflow/dto.py`
- `get_node_spec()` / `all_node_specs()` in `services/integrations/registry.py`
`RFNodeDTO` validates integration nodes by `type` through the registry. That is
the intended extension path.
## Live Call Path
If the integration needs live call data, implement `create_runtime_sessions(...)`
in `runtime.py` and return `IntegrationRuntimeSession` objects.
The generic wiring is already in `api/services/pipecat/run_pipeline.py`:
- `create_runtime_sessions(IntegrationRuntimeContext(...))` is called before the
pipeline task starts.
- Each returned session gets `session.attach(task)` called.
Use this only for lightweight live collection:
- attach task observers
- read context messages
- capture timing / turn / tool events
- build an in-memory snapshot
Do not do outbound network I/O in the live path unless there is a very strong
reason. Prefer the standard pattern: collect live, deliver after the call.
`IntegrationRuntimeContext` gives you:
- `workflow_run_id`
- `workflow_run`
- `workflow_graph`
- `run_definition`
- `user_config`
- `is_realtime`
- `context_messages_provider`
Typical runtime pattern:
- scan `context.workflow_graph.nodes.values()` for enabled nodes of your type
- if none are enabled, return `[]`
- build one collector/session per workflow run, not per node, unless the
integration truly needs multiple independent collectors
## Call-Finish Snapshot Path
`api/services/pipecat/event_handlers.py` finalizes runtime sessions before the
engine is cleaned up.
The generic flow:
1. `on_pipeline_finished` builds `gathered_context`
2. each runtime session gets `await session.on_call_finished(...)`
3. returned dicts are merged into `integration_logs`
4. those logs are persisted into `workflow_run.logs`
Use `on_call_finished(...)` to emit a compact, serializable snapshot that the
post-call completion handler can consume later. Return `None` if there is nothing
to persist.
This is the handoff between the live call path and the post-call task path.
## Post-Call Completion Path
If the integration needs durable artifacts, public URLs, retries, or external
delivery, implement `run_completion(nodes, context)` in `completion.py`.
The generic orchestration is already in `api/tasks/run_integrations.py`:
1. load the pinned workflow definition from the workflow run
2. create a public token if post-call work exists
3. run QA nodes first
4. run registered integration completion handlers
5. run webhook nodes last
Your handler receives:
- `nodes`: raw workflow node dicts for your node types only
- `IntegrationCompletionContext`:
- `workflow_run_id`
- `workflow_run`
- `workflow_definition`
- `definition_id`
- `organization_id`
- `public_token`
Expected completion handler pattern:
- validate each node with `YourNodeData.model_validate(node.get("data", {}))`
- skip disabled nodes
- read any runtime snapshot from `context.workflow_run.logs`
- build durable URLs using `public_token` when appropriate
- perform external delivery
- return a result dict keyed per node, usually with `node_id` embedded
Returned data is merged into `workflow_run.annotations`.
Do not assume completion runs inside the live pipeline process. Treat it as a
separate post-call worker step.
## Optional Routes
If an integration exposes HTTP routes, put them in `routes.py` and include the
router in `IntegrationPackageSpec.routers`.
Routers are mounted automatically by `api/routes/main.py` through `all_routers()`.
Do not edit `routes/main.py` for per-integration route wiring.
## Import Discipline
Keep package import side effects light.
The integration loader runs during:
- node-type/spec enumeration
- tests
- route startup
- registry access
So avoid top-level imports that require environment variables, network access,
or heavyweight initialization when possible. Prefer lazy imports inside
`run_completion()` / `create_runtime_sessions()` if the dependency is optional or
environment-sensitive.
## Testing Expectations
At minimum, new integrations should add coverage for:
- node model validation
- generated spec/example validity
- secret masking + masked round-trip preservation if secrets exist
- runtime snapshot creation if live collectors exist
- completion handler happy path and disabled-node skip path
If you change shared integration machinery, test the framework in the generic
code path, not only the concrete integration.

View file

@ -0,0 +1,39 @@
from api.services.integrations.base import (
IntegrationCompletionContext,
IntegrationNodeRegistration,
IntegrationPackageSpec,
IntegrationRuntimeContext,
IntegrationRuntimeSession,
)
from api.services.integrations.registry import (
all_node_specs,
all_packages,
all_routers,
create_runtime_sessions,
get_node_data_model,
get_node_registration,
get_node_secret_fields,
get_node_spec,
has_completion_handlers,
register_package,
run_completion_handlers,
)
__all__ = [
"IntegrationCompletionContext",
"IntegrationNodeRegistration",
"IntegrationPackageSpec",
"IntegrationRuntimeContext",
"IntegrationRuntimeSession",
"all_node_specs",
"all_packages",
"all_routers",
"create_runtime_sessions",
"get_node_data_model",
"get_node_registration",
"get_node_secret_fields",
"get_node_spec",
"has_completion_handlers",
"register_package",
"run_completion_handlers",
]

View file

@ -0,0 +1,69 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Protocol
from fastapi import APIRouter
from api.services.workflow.node_data import BaseNodeData
from api.services.workflow.node_specs._base import NodeSpec
class IntegrationRuntimeSession(Protocol):
name: str
def attach(self, task: Any) -> None: ...
async def on_call_finished(
self,
*,
gathered_context: dict[str, Any],
) -> dict[str, Any] | None: ...
@dataclass(frozen=True)
class IntegrationRuntimeContext:
workflow_run_id: int
workflow_run: Any
workflow_graph: Any
run_definition: Any
user_config: Any
is_realtime: bool
context_messages_provider: Callable[[], list[dict[str, Any]]]
@dataclass(frozen=True)
class IntegrationCompletionContext:
workflow_run_id: int
workflow_run: Any
workflow_definition: dict[str, Any]
definition_id: int | None
organization_id: int
public_token: str | None
RuntimeFactory = Callable[
[IntegrationRuntimeContext],
list[IntegrationRuntimeSession],
]
CompletionHandler = Callable[
[list[dict[str, Any]], IntegrationCompletionContext],
Awaitable[dict[str, Any]],
]
@dataclass(frozen=True)
class IntegrationNodeRegistration:
type_name: str
data_model: type[BaseNodeData]
node_spec: NodeSpec
sensitive_fields: tuple[str, ...] = ()
@dataclass(frozen=True)
class IntegrationPackageSpec:
name: str
nodes: tuple[IntegrationNodeRegistration, ...] = ()
routers: tuple[APIRouter, ...] = ()
create_runtime_sessions: RuntimeFactory | None = None
run_completion: CompletionHandler | None = None

View file

@ -0,0 +1,21 @@
from __future__ import annotations
import importlib
import pkgutil
_INTERNAL_MODULES = {"base", "loader", "registry"}
_loaded = False
def ensure_integrations_loaded() -> None:
global _loaded
if _loaded:
return
package = importlib.import_module("api.services.integrations")
for module_info in pkgutil.iter_modules(package.__path__):
if module_info.name in _INTERNAL_MODULES:
continue
importlib.import_module(f"{package.__name__}.{module_info.name}")
_loaded = True

View file

@ -1,253 +0,0 @@
import hashlib
import json
import os
from typing import Any, Dict
import httpx
from fastapi import HTTPException
from loguru import logger
from pydantic import BaseModel
from api.db import db_client
NANGO_ALLOWED_INTEGRATIONS = [
i.strip() for i in os.environ.get("NANGO_ALLOWED_INTEGRATIONS", "slack").split(",")
]
class NangoWebhookRequest(BaseModel):
type: str
connectionId: str
providerConfigKey: str
authMode: str
provider: str
environment: str
operation: str
endUser: dict # Contains endUserId and organizationId
success: bool
class NangoService:
def __init__(self):
self.base_url = "https://api.nango.dev"
self.secret_key = os.getenv("NANGO_API_KEY")
def _verify_webhook_signature(
self, request_body: str, signature: str = None
) -> bool:
"""
Verify the webhook signature using SHA256 hash.
Args:
request_body: The raw request body as string
signature: The signature from request headers (optional for now)
Returns:
True if signature is valid
"""
expected_signature = self.secret_key + request_body
expected_hash = hashlib.sha256(expected_signature.encode("utf-8")).hexdigest()
return expected_hash == signature
async def create_session(
self, user_id: str, organization_id: int
) -> Dict[str, Any]:
"""
Create a Nango session for the given user and organization.
Args:
user_id: The end user ID
organization_id: The organization ID
Returns:
Response from Nango API
"""
if not self.secret_key:
raise ValueError("NANGO_SECRET_KEY environment variable is not set")
headers = {
"Authorization": f"Bearer {self.secret_key}",
"Content-Type": "application/json",
}
payload = {
"end_user": {"id": user_id},
"organization": {"id": str(organization_id)},
"allowed_integrations": NANGO_ALLOWED_INTEGRATIONS,
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/connect/sessions", headers=headers, json=payload
)
if response.status_code != 201:
raise httpx.HTTPStatusError(
f"Nango API error: {response.status_code}",
request=response.request,
response=response,
)
return response.json()
async def process_webhook(
self, raw_body: bytes, signature: str = None
) -> Dict[str, str]:
"""
Process incoming Nango webhook request.
Args:
raw_body: The raw request body as bytes
signature: Optional signature from request headers
Returns:
Dict with status and message
"""
# Decode and parse the request body
try:
body_text = raw_body.decode("utf-8")
webhook_json = json.loads(body_text) if body_text else {}
logger.debug(f"received webhook from nango: {webhook_json}")
except json.JSONDecodeError as e:
logger.error(f"JSON decode error: {e} body_text: {body_text}")
raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(e)}")
# Verify webhook signature
if not self._verify_webhook_signature(body_text, signature):
raise HTTPException(status_code=401, detail="Invalid webhook signature")
# Parse webhook data
try:
webhook_data = NangoWebhookRequest(**webhook_json)
except Exception as e:
logger.error(f"Failed to parse webhook data: {e}")
raise HTTPException(
status_code=400, detail=f"Invalid webhook format: {str(e)}"
)
# Extract user and organization IDs from the webhook payload
end_user = webhook_data.endUser
if (
not end_user
or "endUserId" not in end_user
or "organizationId" not in end_user
):
raise HTTPException(
status_code=400, detail="Missing endUser information in webhook payload"
)
user_id = int(end_user["endUserId"])
organization_id = int(end_user["organizationId"])
# Use the connectionId as the integration_id since it's unique per integration
integration_id = webhook_data.connectionId
# Initialize connection_details
connection_details = {}
# Fetch connection details if type is auth and provider is slack
if webhook_data.type == "auth":
connection_details = await self._fetch_connection_details(
integration_id, webhook_data.provider
)
# Create the integration in the database
integration = await db_client.create_integration(
integration_id=integration_id,
organisation_id=organization_id,
provider=webhook_data.provider,
created_by=user_id,
is_active=True,
connection_details=connection_details,
)
return {
"status": "success",
"message": f"Integration created successfully with ID: {integration.id}",
}
async def _fetch_connection_details(
self, connection_id: str, provider_key: str
) -> Dict[str, Any]:
"""
Fetch connection details from Nango API for a given connection ID.
Args:
connection_id: The connection ID from the webhook
Returns:
Connection details as a dictionary
"""
headers = {
"Authorization": f"Bearer {self.secret_key}",
"Content-Type": "application/json",
}
url = f"{self.base_url}/connection/{connection_id}/?provider_config_key={provider_key}"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers)
if response.status_code != 200:
logger.error(
f"Failed to fetch connection details: {response.status_code} - {response.text}"
)
raise httpx.HTTPStatusError(
f"Nango API error while fetching connection: {response.status_code}",
request=response.request,
response=response,
)
connection_details = response.json()
return connection_details
except httpx.HTTPError as e:
logger.error(f"HTTP error while fetching connection details: {e}")
# Return empty dict if API call fails, but log the error
return {}
async def get_access_token(
self, connection_id: str, provider_config_key: str
) -> Dict[str, Any]:
"""
Get the latest access token for a connection from Nango.
Args:
connection_id: The connection ID
provider_config_key: The provider config key (e.g., 'google-sheet')
Returns:
Dict containing access token and other connection details
"""
headers = {
"Authorization": f"Bearer {self.secret_key}",
"Content-Type": "application/json",
}
url = f"{self.base_url}/connection/{connection_id}?provider_config_key={provider_config_key}"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers)
if response.status_code != 200:
logger.error(
f"Failed to get access token: {response.status_code} - {response.text}"
)
raise httpx.HTTPStatusError(
f"Nango API error: {response.status_code}",
request=response.request,
response=response,
)
return response.json()
except httpx.HTTPError as e:
logger.error(f"HTTP error while getting access token: {e}")
raise
# Create a singleton instance
nango_service = NangoService()

View file

@ -0,0 +1,128 @@
from __future__ import annotations
from typing import Any
from api.services.integrations.base import (
IntegrationCompletionContext,
IntegrationNodeRegistration,
IntegrationPackageSpec,
IntegrationRuntimeContext,
)
from api.services.workflow.node_data import BaseNodeData
_PACKAGE_REGISTRY: dict[str, IntegrationPackageSpec] = {}
def register_package(spec: IntegrationPackageSpec) -> IntegrationPackageSpec:
existing = _PACKAGE_REGISTRY.get(spec.name)
if existing is not None and existing is not spec:
raise ValueError(
f"Duplicate integration package registration for {spec.name!r}"
)
_PACKAGE_REGISTRY[spec.name] = spec
return spec
def _ensure_loaded() -> None:
from api.services.integrations.loader import ensure_integrations_loaded
ensure_integrations_loaded()
def all_packages() -> list[IntegrationPackageSpec]:
_ensure_loaded()
return [_PACKAGE_REGISTRY[name] for name in sorted(_PACKAGE_REGISTRY)]
def get_package(name: str) -> IntegrationPackageSpec | None:
_ensure_loaded()
return _PACKAGE_REGISTRY.get(name)
def get_node_registration(type_name: str) -> IntegrationNodeRegistration | None:
_ensure_loaded()
for package in _PACKAGE_REGISTRY.values():
for node in package.nodes:
if node.type_name == type_name:
return node
return None
def get_node_data_model(type_name: str) -> type[BaseNodeData] | None:
registration = get_node_registration(type_name)
return registration.data_model if registration else None
def get_node_spec(type_name: str):
registration = get_node_registration(type_name)
return registration.node_spec if registration else None
def get_node_secret_fields(type_name: str) -> tuple[str, ...]:
registration = get_node_registration(type_name)
return registration.sensitive_fields if registration else ()
def all_node_specs():
_ensure_loaded()
specs = []
for package in all_packages():
specs.extend(node.node_spec for node in package.nodes)
return specs
def all_routers():
_ensure_loaded()
routers = []
for package in all_packages():
routers.extend(package.routers)
return routers
def create_runtime_sessions(
context: IntegrationRuntimeContext,
):
_ensure_loaded()
sessions = []
for package in all_packages():
if package.create_runtime_sessions is None:
continue
sessions.extend(package.create_runtime_sessions(context))
return sessions
def iter_completion_packages(
workflow_definition: dict[str, Any],
):
_ensure_loaded()
nodes = workflow_definition.get("nodes", []) if workflow_definition else []
for package in all_packages():
node_types = {node.type_name for node in package.nodes}
package_nodes = [
node
for node in nodes
if isinstance(node, dict) and node.get("type") in node_types
]
if package_nodes:
yield package, package_nodes
def has_completion_handlers(workflow_definition: dict[str, Any]) -> bool:
return any(
package.run_completion is not None
for package, _nodes in iter_completion_packages(workflow_definition)
)
async def run_completion_handlers(
*,
context: IntegrationCompletionContext,
) -> dict[str, Any]:
results: dict[str, Any] = {}
for package, nodes in iter_completion_packages(context.workflow_definition):
if package.run_completion is None:
continue
package_result = await package.run_completion(nodes, context)
if package_result:
results.update(package_result)
return results

View file

@ -0,0 +1,19 @@
from __future__ import annotations
from api.services.integrations.base import IntegrationPackageSpec
from api.services.integrations.registry import register_package
from .completion import run_completion
from .node import NODE
from .runtime import create_runtime_sessions
PACKAGE = register_package(
IntegrationPackageSpec(
name="tuner",
nodes=(NODE,),
create_runtime_sessions=create_runtime_sessions,
run_completion=run_completion,
)
)
__all__ = ["PACKAGE"]

View file

@ -0,0 +1,71 @@
from __future__ import annotations
from typing import Any
import httpx
from loguru import logger
from pydantic import BaseModel, field_validator
class TunerDeliveryConfig(BaseModel):
base_url: str
api_key: str
workspace_id: int
agent_id: str
@field_validator("api_key", "agent_id")
@classmethod
def _must_not_be_empty(cls, value: str) -> str:
if not value or not value.strip():
raise ValueError("must not be empty")
return value
@field_validator("workspace_id")
@classmethod
def _workspace_must_be_positive(cls, value: int) -> int:
if value <= 0:
raise ValueError("must be a positive integer")
return value
async def post_call(
config: TunerDeliveryConfig,
payload: dict[str, Any],
) -> dict[str, Any]:
url = (
f"{config.base_url}/api/v1/public/call"
f"?workspace_id={config.workspace_id}"
f"&agent_remote_identifier={config.agent_id}"
)
headers = {"Authorization": f"Bearer {config.api_key}"}
logger.info(
"[tuner] posting completed call {} to workspace {} / agent {}",
payload.get("call_id"),
config.workspace_id,
config.agent_id,
)
async with httpx.AsyncClient(timeout=10) as client:
response = await client.post(url, json=payload, headers=headers)
if response.status_code == 409:
logger.info("[tuner] call {} already exists in tuner", payload.get("call_id"))
return {"status": "duplicate", "status_code": response.status_code}
if response.status_code >= 400:
logger.error(
"[tuner] POST failed for call {} with status {}: {}",
payload.get("call_id"),
response.status_code,
response.text[:200],
)
response.raise_for_status()
logger.info(
"[tuner] POST succeeded for call {} with status {}",
payload.get("call_id"),
response.status_code,
)
return {"status": "delivered", "status_code": response.status_code}

View file

@ -0,0 +1,191 @@
from __future__ import annotations
import time
from collections import deque
from dataclasses import dataclass
from typing import Any, Callable
from loguru import logger
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
MetricsFrame,
StartFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.processors.frame_processor import FrameDirection
from pipecat.utils.context.message_sanitization import strip_thought_ids_from_messages
from tuner_pipecat_sdk.accumulator import CallAccumulator
from tuner_pipecat_sdk.payload_builder import build_payload
from api.enums import WorkflowRunMode
TUNER_RECORDING_PLACEHOLDER = "pipecat://no-recording"
@dataclass(frozen=True)
class _PayloadConfig:
call_id: str
call_type: str
recording_url: str
asr_model: str
llm_model: str
tts_model: str
sip_call_id: str | None = None
sip_headers: dict[str, str] | None = None
agent_version: int | None = None
def mode_to_tuner_call_type(mode: str | None) -> str:
if mode in {
WorkflowRunMode.WEBRTC.value,
WorkflowRunMode.SMALLWEBRTC.value,
}:
return "web_call"
return "phone_call"
class TunerCollector(BaseObserver):
"""Collect runtime call metadata and build a deferred Tuner payload."""
def __init__(
self,
*,
workflow_run_id: int,
call_type: str,
asr_model: str = "",
llm_model: str = "",
tts_model: str = "",
agent_version: int | None = None,
max_frames: int = 500,
) -> None:
super().__init__()
self._call_id = str(workflow_run_id)
self._call_type = call_type
self._asr_model = asr_model
self._llm_model = llm_model
self._tts_model = tts_model
self._agent_version = agent_version
self._acc = CallAccumulator()
self._acc.call_start_abs_ns = time.time_ns()
self._pipeline_start_rel_ns: int | None = None
self._context_provider: Callable[[], list[dict[str, Any]]] | None = None
self._processed_frames: set[int] = set()
self._frame_history: deque[int] = deque(maxlen=max_frames)
def attach_context(self, provider: Callable[[], list[dict[str, Any]]]) -> None:
self._context_provider = provider
def set_disconnection_reason(self, reason: str | None) -> None:
if reason:
self._acc.set_disconnection_reason(reason)
def attach_turn_tracking_observer(
self, turn_tracker: TurnTrackingObserver | None
) -> None:
if turn_tracker is None:
return
@turn_tracker.event_handler("on_turn_started")
async def _on_turn_started(_tracker: Any, turn_number: int) -> None:
self._acc.on_turn_started(turn_number, time.time_ns())
@turn_tracker.event_handler("on_turn_ended")
async def _on_turn_ended(
_tracker: Any, turn_number: int, _duration: float, was_interrupted: bool
) -> None:
self._acc.on_turn_ended(turn_number, was_interrupted)
def attach_latency_observer(
self, latency_observer: UserBotLatencyObserver | None
) -> None:
if latency_observer is None:
return
@latency_observer.event_handler("on_latency_measured")
async def _on_latency_measured(_observer: Any, latency: float) -> None:
self._acc.on_latency_measured(latency)
@latency_observer.event_handler("on_latency_breakdown")
async def _on_latency_breakdown(_observer: Any, breakdown: Any) -> None:
self._acc.on_latency_breakdown(breakdown)
async def on_push_frame(self, data: FramePushed):
if data.direction != FrameDirection.DOWNSTREAM:
return
if data.frame.id in self._processed_frames:
return
self._processed_frames.add(data.frame.id)
self._frame_history.append(data.frame.id)
if len(self._processed_frames) > len(self._frame_history):
self._processed_frames = set(self._frame_history)
frame = data.frame
# data.timestamp is a pipeline-relative clock (ns since pipeline start).
# Convert to absolute ns so the accumulator's _rel_ms() works correctly.
if self._pipeline_start_rel_ns is None:
self._pipeline_start_rel_ns = data.timestamp
timestamp_ns = self._acc.call_start_abs_ns + (
data.timestamp - self._pipeline_start_rel_ns
)
if isinstance(frame, StartFrame):
self._acc.on_start(timestamp_ns)
elif isinstance(frame, FunctionCallInProgressFrame):
self._acc.on_function_call_in_progress(frame, timestamp_ns)
elif isinstance(frame, FunctionCallResultFrame):
self._acc.on_function_call_result(frame.tool_call_id, timestamp_ns)
elif isinstance(frame, MetricsFrame):
self._acc.on_metrics_frame(frame)
elif isinstance(frame, UserStartedSpeakingFrame):
self._acc.on_user_started_speaking(timestamp_ns)
elif isinstance(frame, UserStoppedSpeakingFrame):
self._acc.on_user_stopped_speaking(timestamp_ns)
self._acc.on_user_turn_stopped(timestamp_ns)
elif isinstance(frame, BotStartedSpeakingFrame):
self._acc.on_bot_started_speaking(timestamp_ns)
elif isinstance(frame, BotStoppedSpeakingFrame):
self._acc.on_bot_stopped(timestamp_ns)
elif isinstance(frame, VADUserStoppedSpeakingFrame):
self._acc.on_vad_stopped(timestamp_ns)
elif isinstance(frame, (CancelFrame, EndFrame)):
self._acc.on_call_end(timestamp_ns)
def build_payload_snapshot(
self,
*,
recording_url: str = TUNER_RECORDING_PLACEHOLDER,
) -> dict[str, Any] | None:
if self._context_provider is None:
logger.warning(
"[tuner] no context provider attached; skipping payload snapshot"
)
return None
transcript = strip_thought_ids_from_messages(list(self._context_provider()))
payload = build_payload(
self._acc,
_PayloadConfig(
call_id=self._call_id,
call_type=self._call_type,
recording_url=recording_url,
asr_model=self._asr_model,
llm_model=self._llm_model,
tts_model=self._tts_model,
agent_version=self._agent_version,
),
transcript,
)
return payload.to_dict()

View file

@ -0,0 +1,76 @@
from __future__ import annotations
import copy
from datetime import UTC, datetime
from typing import Any
from loguru import logger
from api.constants import BACKEND_API_ENDPOINT, TUNER_BASE_URL
from api.services.integrations.base import IntegrationCompletionContext
from .client import TunerDeliveryConfig, post_call
from .collector import TUNER_RECORDING_PLACEHOLDER
from .node import TunerNodeData
def _build_recording_url(
context: IntegrationCompletionContext,
) -> str | None:
workflow_run = context.workflow_run
if context.public_token:
base_url = f"{BACKEND_API_ENDPOINT}/api/v1/public/download/workflow/{context.public_token}"
return f"{base_url}/recording" if workflow_run.recording_url else None
return workflow_run.recording_url
async def run_completion(
nodes: list[dict[str, Any]],
context: IntegrationCompletionContext,
) -> dict[str, Any]:
results: dict[str, Any] = {}
payload_snapshot = (context.workflow_run.logs or {}).get("tuner_payload")
recording_url = _build_recording_url(context) or TUNER_RECORDING_PLACEHOLDER
for node in nodes:
node_id = node.get("id", "unknown")
try:
tuner_data = TunerNodeData.model_validate(node.get("data", {}))
except Exception as exc:
logger.warning(f"Tuner node #{node_id} failed validation, skipping: {exc}")
results[f"tuner_{node_id}"] = {"error": "validation_failed"}
continue
if not tuner_data.tuner_enabled:
logger.debug(f"Tuner node '{tuner_data.name}' is disabled, skipping")
continue
if not payload_snapshot:
logger.warning(
f"Tuner payload snapshot missing for node '{tuner_data.name}' (#{node_id})"
)
results[f"tuner_{node_id}"] = {"error": "missing_payload_snapshot"}
continue
payload = copy.deepcopy(payload_snapshot)
payload["recording_url"] = recording_url
try:
config = TunerDeliveryConfig(
base_url=TUNER_BASE_URL,
api_key=tuner_data.tuner_api_key or "",
workspace_id=tuner_data.tuner_workspace_id or 0,
agent_id=tuner_data.tuner_agent_id or "",
)
delivery = await post_call(config, payload)
results[f"tuner_{node_id}"] = {
**delivery,
"workspace_id": tuner_data.tuner_workspace_id,
"agent_id": tuner_data.tuner_agent_id,
"exported_at": datetime.now(UTC).isoformat(),
}
except Exception as exc:
logger.error(f"Tuner export failed for node '{tuner_data.name}': {exc}")
results[f"tuner_{node_id}"] = {"error": str(exc)}
return results

View file

@ -0,0 +1,139 @@
from __future__ import annotations
from pydantic import model_validator
from api.services.integrations.base import IntegrationNodeRegistration
from api.services.workflow.node_data import BaseNodeData
from api.services.workflow.node_specs._base import (
GraphConstraints,
NodeCategory,
NodeExample,
PropertyType,
)
from api.services.workflow.node_specs.model_spec import (
build_spec,
node_spec,
spec_field,
)
@node_spec(
name="tuner",
display_name="Tuner",
description="Export the completed call to Tuner for Agent Observability",
llm_hint=(
"Tuner is a post-call observability export. It does not participate in the "
"conversation graph and should not be connected to other nodes."
),
category=NodeCategory.integration,
icon="Activity",
examples=[
NodeExample(
name="tuner_export",
data={
"name": "Primary Tuner Export",
"tuner_enabled": True,
"tuner_agent_id": "sales-bot-prod",
"tuner_workspace_id": 42,
"tuner_api_key": "tuner_live_xxxxxxxx",
},
)
],
graph_constraints=GraphConstraints(
min_incoming=0,
max_incoming=0,
min_outgoing=0,
max_outgoing=0,
),
property_order=(
"name",
"tuner_enabled",
"tuner_agent_id",
"tuner_workspace_id",
"tuner_api_key",
),
field_overrides={
"name": {
"spec_default": "Tuner",
"description": "Short identifier for this Tuner export configuration.",
},
"tuner_enabled": {
"display_name": "Enabled",
"description": "When false, Dograh skips exporting this call to Tuner.",
},
"tuner_agent_id": {
"display_name": "Tuner Agent ID",
"description": "The agent identifier registered in your Tuner workspace.",
"required": True,
},
"tuner_workspace_id": {
"display_name": "Tuner Workspace ID",
"description": "Your numeric Tuner workspace ID.",
"required": True,
"min_value": 1,
},
"tuner_api_key": {
"display_name": "Tuner API Key",
"description": "Bearer token used when posting completed calls to Tuner.",
"required": True,
},
},
)
class TunerNodeData(BaseNodeData):
tuner_enabled: bool = spec_field(
default=True,
ui_type=PropertyType.boolean,
display_name="Enabled",
description="When false, Dograh skips exporting this call to Tuner.",
)
tuner_agent_id: str | None = spec_field(
default=None,
ui_type=PropertyType.string,
display_name="Tuner Agent ID",
description="The agent identifier registered in your Tuner workspace.",
)
tuner_workspace_id: int | None = spec_field(
default=None,
gt=0,
ui_type=PropertyType.number,
display_name="Tuner Workspace ID",
description="Your numeric Tuner workspace ID.",
)
tuner_api_key: str | None = spec_field(
default=None,
ui_type=PropertyType.string,
display_name="Tuner API Key",
description="Bearer token used when posting completed calls to Tuner.",
)
@model_validator(mode="after")
def _validate_enabled_config(self):
if not self.tuner_enabled:
return self
missing: list[str] = []
if not self.tuner_agent_id or not self.tuner_agent_id.strip():
missing.append("tuner_agent_id")
if self.tuner_workspace_id is None:
missing.append("tuner_workspace_id")
if not self.tuner_api_key or not self.tuner_api_key.strip():
missing.append("tuner_api_key")
if missing:
fields = ", ".join(missing)
raise ValueError(
f"Tuner node is enabled but missing required fields: {fields}"
)
return self
SPEC = build_spec(TunerNodeData)
NODE = IntegrationNodeRegistration(
type_name="tuner",
data_model=TunerNodeData,
node_spec=SPEC,
sensitive_fields=("tuner_api_key",),
)

View file

@ -0,0 +1,101 @@
from __future__ import annotations
from typing import Any
from api.services.configuration.registry import ServiceProviders
from api.services.integrations.base import (
IntegrationRuntimeContext,
IntegrationRuntimeSession,
)
from .collector import TunerCollector, mode_to_tuner_call_type
def _format_model_label(provider: str | None, model: str | None) -> str:
if provider and model:
return f"{provider}/{model}"
if model:
return model
return provider or ""
def _resolve_model_labels(context: IntegrationRuntimeContext) -> tuple[str, str, str]:
user_config = context.user_config
if context.is_realtime and user_config.realtime:
realtime_provider = user_config.realtime.provider
realtime_model = user_config.realtime.model
llm_model = _format_model_label(realtime_provider, realtime_model)
if realtime_provider in {
ServiceProviders.GOOGLE_REALTIME.value,
ServiceProviders.GOOGLE_VERTEX_REALTIME.value,
ServiceProviders.OPENAI_REALTIME.value,
}:
return "", llm_model, ""
return "", llm_model, ""
return (
_format_model_label(
getattr(user_config.stt, "provider", None),
getattr(user_config.stt, "model", None),
),
_format_model_label(
getattr(user_config.llm, "provider", None),
getattr(user_config.llm, "model", None),
),
_format_model_label(
getattr(user_config.tts, "provider", None),
getattr(user_config.tts, "model", None),
),
)
class TunerRuntimeSession(IntegrationRuntimeSession):
name = "tuner"
def __init__(self, collector: TunerCollector) -> None:
self._collector = collector
def attach(self, task: Any) -> None:
self._collector.attach_turn_tracking_observer(task.turn_tracking_observer)
self._collector.attach_latency_observer(task.user_bot_latency_observer)
task.add_observer(self._collector)
async def on_call_finished(
self,
*,
gathered_context: dict[str, Any],
) -> dict[str, Any] | None:
self._collector.set_disconnection_reason(
gathered_context.get("call_disposition")
)
payload = self._collector.build_payload_snapshot()
if payload is None:
return None
return {"tuner_payload": payload}
def create_runtime_sessions(
context: IntegrationRuntimeContext,
) -> list[IntegrationRuntimeSession]:
tuner_nodes = [
node
for node in context.workflow_graph.nodes.values()
if node.node_type == "tuner" and getattr(node.data, "tuner_enabled", True)
]
if not tuner_nodes:
return []
asr_model, llm_model, tts_model = _resolve_model_labels(context)
collector = TunerCollector(
workflow_run_id=context.workflow_run_id,
call_type=mode_to_tuner_call_type(context.workflow_run.mode),
asr_model=asr_model,
llm_model=llm_model,
tts_model=tts_model,
agent_version=getattr(context.run_definition, "version_number", None),
)
collector.attach_context(context.context_messages_provider)
return [TunerRuntimeSession(collector)]

View file

@ -5,6 +5,7 @@ from loguru import logger
from api.db import db_client
from api.enums import PostHogEvent, WorkflowRunState
from api.services.campaign.circuit_breaker import circuit_breaker
from api.services.integrations import IntegrationRuntimeSession
from api.services.pipecat.audio_config import AudioConfig
from api.services.pipecat.audio_playback import play_audio_loop
from api.services.pipecat.in_memory_buffers import (
@ -67,6 +68,7 @@ def register_event_handlers(
audio_config=AudioConfig,
pre_call_fetch_task: asyncio.Task | None = None,
user_provider_id: str | None = None,
integration_runtime_sessions: list[IntegrationRuntimeSession] | None = None,
):
"""Register all event handlers for transport and task events.
@ -272,6 +274,20 @@ def register_event_handlers(
)
# Clean up engine resources (including voicemail detector)
integration_logs: dict[str, object] = {}
for runtime_session in integration_runtime_sessions or []:
try:
session_logs = await runtime_session.on_call_finished(
gathered_context=gathered_context
)
if session_logs:
integration_logs.update(session_logs)
except Exception as e:
logger.error(
f"Error finalizing integration runtime session '{runtime_session.name}': {e}",
exc_info=True,
)
await engine.cleanup()
# ------------------------------------------------------------------
@ -321,14 +337,11 @@ def register_event_handlers(
)
)
# Save real-time feedback logs to workflow run
logs_update: dict[str, object] = {}
if not in_memory_logs_buffer.is_empty:
try:
feedback_events = in_memory_logs_buffer.get_events()
await db_client.update_workflow_run(
run_id=workflow_run_id,
logs={"realtime_feedback_events": feedback_events},
)
logs_update["realtime_feedback_events"] = feedback_events
logger.debug(
f"Saved {len(feedback_events)} feedback events to workflow run logs"
)
@ -337,6 +350,17 @@ def register_event_handlers(
else:
logger.debug("Logs buffer is empty, skipping save")
logs_update.update(integration_logs)
if logs_update:
try:
await db_client.update_workflow_run(
run_id=workflow_run_id,
logs=logs_update,
)
except Exception as e:
logger.error(f"Error saving workflow run logs: {e}", exc_info=True)
# Write buffers to temp files and enqueue combined processing task
audio_temp_path = None
transcript_temp_path = None

View file

@ -7,6 +7,10 @@ from loguru import logger
from api.db import db_client
from api.enums import WorkflowRunMode
from api.services.configuration.registry import ServiceProviders
from api.services.integrations import (
IntegrationRuntimeContext,
create_runtime_sessions,
)
from api.services.pipecat.audio_config import AudioConfig, create_audio_config
from api.services.pipecat.event_handlers import (
register_audio_data_handler,
@ -525,6 +529,18 @@ async def _run_pipeline(
# Create pipeline components
audio_buffer, context = create_pipeline_components(audio_config)
integration_runtime_sessions = create_runtime_sessions(
IntegrationRuntimeContext(
workflow_run_id=workflow_run_id,
workflow_run=workflow_run,
workflow_graph=workflow_graph,
run_definition=run_definition,
user_config=user_config,
is_realtime=is_realtime,
context_messages_provider=lambda: context.messages,
)
)
# Set the context, audio_config, and audio_buffer after creation
engine.set_context(context)
engine.set_audio_config(audio_config)
@ -717,6 +733,14 @@ async def _run_pipeline(
# Create pipeline task with audio configuration
task = create_pipeline_task(pipeline, workflow_run_id, audio_config)
for runtime_session in integration_runtime_sessions:
runtime_session.attach(task)
logger.info(
"[integrations] attached runtime session '{}' for workflow run {}",
runtime_session.name,
workflow_run_id,
)
# Now set the task and transport output on the engine
engine.set_task(task)
engine.set_transport_output(transport.output())
@ -780,6 +804,7 @@ async def _run_pipeline(
audio_config=audio_config,
pre_call_fetch_task=pre_call_fetch_task,
user_provider_id=user_provider_id,
integration_runtime_sessions=integration_runtime_sessions,
)
register_audio_data_handler(audio_buffer, workflow_run_id, in_memory_audio_buffer)

View file

@ -0,0 +1,10 @@
# Telephony
Shared telephony code lives here. Provider-specific code lives in `providers/`;
read `providers/AGENTS.md` before changing a provider package.
- Keep cross-provider contracts, registry/factory wiring, shared status/transfer handling, and org-scoped config resolution in this folder.
- Keep provider-specific transports, serializers, config models, and webhook handlers in `providers/`.
- Resolve providers through the shared telephony helpers in this folder; do not instantiate provider classes directly from routes, tasks, or unrelated services.
- Keep telephony config lookups tenant-safe and respect any run-scoped telephony configuration carried on a workflow run.
- Keep provider-specific HTTP routes in provider packages; shared route glue belongs in `api/routes/`.

View file

@ -0,0 +1,123 @@
# Telephony Providers
Each subdirectory here is a self-registering telephony provider. Adding a new one should touch this folder plus **exactly two lines** outside it. If a change you're making requires editing `factory.py`, `audio_config.py`, `run_pipeline.py`, `routes/telephony.py`, or any frontend file, stop — that's a smell. Push the variation through the registry instead.
## Anatomy of a provider package
```
providers/<name>/
├── __init__.py # Required. Builds + register()s ProviderSpec
├── config.py # Required. Pydantic Request + Response, both with `provider: Literal["<name>"]`
├── provider.py # Required. TelephonyProvider subclass
├── transport.py # Required. async create_transport(...) -> FastAPIWebsocketTransport
├── serializers.py # Optional but conventional. Re-export from pipecat
├── routes.py # Optional. APIRouter mounted lazily under /api/v1/telephony
└── strategies.py # Optional. Transfer/Hangup strategies for the frame serializer
```
Every file is provider-local. Nothing here imports another provider package.
## The two edits outside this folder
After creating `providers/<name>/`:
1. `providers/__init__.py` — add `<name>` to the import-for-side-effects list. Registration runs at import time.
2. `api/schemas/telephony_config.py` — import `<Name>ConfigurationRequest`/`Response` and add the request to the `TelephonyConfigRequest` `Union[...]` and the response as an optional field on `TelephonyConfigurationResponse`.
If you find yourself editing anything else, re-read the registry plumbing first:
| Want to change... | Source of truth |
| ------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| Outbound provider lookup | `factory.get_default_telephony_provider`, `get_telephony_provider_by_id`, and `get_telephony_provider_for_run` read `registry.get(name).provider_cls` |
| Stored credentials → constructor dict | `ProviderSpec.config_loader` |
| Audio sample rate / VAD rate | `ProviderSpec.transport_sample_rate` (full `AudioConfig` is built in `pipecat/audio_config.py::create_audio_config`) |
| Which transport runs in `run_pipeline_telephony` | `ProviderSpec.transport_factory` |
| Save-request validation + masked response shape | `ProviderSpec.config_request_cls` / `config_response_cls` |
| Form rendered by the telephony-config UI | `ProviderSpec.ui_metadata` (`ProviderUIField` list) |
| Which credential masks on read | `ui_metadata.fields[*].sensitive=True` (no separate list) |
| Inbound webhook → config row matching | `ProviderSpec.account_id_credential_field` |
| HTTP routes (answer URL, status callbacks) | `providers/<name>/routes.py` (auto-mounted via `importlib`) |
## ProviderSpec — minimum viable shape
```python
SPEC = ProviderSpec(
name="<name>", # registry key, WorkflowRunMode value, stored discriminator
provider_cls=YourProvider,
config_loader=_config_loader, # raw dict from DB → constructor dict
transport_factory=create_transport,
transport_sample_rate=8000, # wire-format rate; pipecat derives the full AudioConfig
config_request_cls=YourProviderConfigurationRequest,
config_response_cls=YourProviderConfigurationResponse,
ui_metadata=ProviderUIMetadata(...), # drives the form UI
account_id_credential_field="api_key", # "" if provider has no account-id concept
)
register(SPEC)
```
`ProviderSpec` is frozen — immutable post-registration. Re-registration with the same instance is a no-op; re-registration with a different instance raises.
## Registration is import-driven, not config-driven
`api/services/telephony/__init__.py` imports `providers/` for side effects. Don't add a registration call elsewhere — by the time `factory`, `audio_config`, or `run_pipeline_telephony` look the spec up, the package init has already executed.
The package init **does not import `routes.py`**`api/routes/telephony.py::_mount_provider_routers()` walks `registry.all_specs()` and uses `importlib.import_module(f"...providers.{spec.name}.routes")`, treating `ModuleNotFoundError` as "no routes for this provider." This is what keeps `from api.services.telephony.base import TelephonyProvider` from fanning out to every route handler in the app. Don't undo it by importing `.routes` from `__init__.py`.
## Conventions
### `provider: Literal["<name>"]` on both Request and Response
Pydantic's discriminated union dispatches on this field. Forgetting `Literal` makes the union accept any provider's payload as yours. Default it to the literal so save calls don't have to send it explicitly.
### Transports load credentials lazily
Always:
```python
from api.services.telephony.factory import load_credentials_for_transport
config = await load_credentials_for_transport(
organization_id, telephony_configuration_id, expected_provider="<name>",
)
```
Never read the org's default config from `transport.py`. The workflow run carries `telephony_configuration_id` in `initial_context` for multi-config orgs; `load_credentials_for_transport` resolves the right row and validates the provider matches.
### `_config_loader` is a pure dict reshape
It runs over `TelephonyConfigurationModel.credentials` (the JSONB column). Don't do I/O in it. Don't pull `from_numbers` from credentials — the factory attaches active phone numbers from `telephony_phone_numbers` after the loader runs, by joining and normalizing addresses.
### Sensitive fields
Mark every credential field `sensitive=True` in `ProviderUIMetadata`. The org routes derive masking from `ui_metadata`, not from a separate hardcoded list. If you re-submit a masked value, `preserve_masked_fields` restores the original — relying on this means you should never write `sensitive=False` on a real secret to "make the form simpler."
### Inbound webhook routing
When multiple configs of the same provider live in one org (e.g. two Twilio sub-accounts), the inbound dispatcher matches the webhook to a config by `credentials[<account_id_credential_field>]`. Set this to whatever your provider stamps on inbound payloads (`account_sid` for Twilio, `auth_id` for Plivo, etc.). Set `""` only when the provider truly has no account-id concept (e.g. ARI — there's at most one config per org).
### `configure_inbound` defaults to no-op
Override only when the provider supports programmatic webhook binding (Plivo `application_id`, Telnyx app config). Markup-response providers that learn the webhook URL from console-side configuration leave the default. Returning `ProviderSyncResult(ok=False, message="...")` surfaces a non-fatal warning to the user without aborting the DB write.
## Reference implementations
Pick the closest shape and copy from it.
| Provider | Pick when... |
| ------------ | ------------------------------------------------------------------------------------------------------------------------------ |
| `twilio/` | Markup-response (TwiML), HMAC-signed webhooks, conference-style transfers, status callbacks. The most full-featured reference. |
| `plivo/` | Markup-response with multi-callback signature schemes, programmatic answer-URL sync via Application API. |
| `vonage/` | JWT auth, 16 kHz Linear PCM wire format, NCCO JSON responses. |
| `cloudonix/` | SIP-trunk-style with custom transfer/hangup strategies. |
| `telnyx/` | Call-control style — REST calls to answer/stream rather than markup response. |
| `vobiz/` | Body-signed webhooks (signature covers raw bytes). |
| `ari/` | Smallest viable: no `routes.py`, no `verify_inbound_signature`, WebSocket-only, no account-id. |
## What NOT to do
- **Don't import another provider's `provider.py` or `transport.py`.** Cross-provider behavior belongs in `services/telephony/` (e.g. `status_processor`, `ari_manager`, `call_transfer_manager`), not in another provider's package.
- **Don't add a hardcoded provider list anywhere.** If you need to iterate, use `registry.all_specs()` / `registry.names()`.
- **Don't add a route under `routes/telephony.py` for a single provider.** Provider-specific handlers go in `providers/<name>/routes.py`. Cross-provider handlers (`/inbound/run`, `/twiml`) stay in `routes/telephony.py`.
- **Don't import `.routes` from a provider's `__init__.py`.** That's the cycle we deliberately broke — see "Registration is import-driven."
- **Don't write a frontend form for a new provider.** The UI consumes `GET /api/v1/organizations/telephony-providers/metadata` and renders generically from `ProviderUIField`. If a `field.type` you need doesn't exist (`text`/`password`/`textarea`/`string-array`/`number`), extend the shared telephony UI under `ui/src/components/telephony/` once — not per provider.
- **Don't run a database migration to add a provider.** The discriminator lives in JSONB credentials and a `VARCHAR(64)` `mode` column; nothing in the DB schema knows the set of provider names.

View file

@ -1,123 +1 @@
# Telephony Providers
Each subdirectory here is a self-registering telephony provider. Adding a new one should touch this folder plus **exactly two lines** outside it. If a change you're making requires editing `factory.py`, `audio_config.py`, `run_pipeline.py`, `routes/telephony.py`, or any frontend file, stop — that's a smell. Push the variation through the registry instead.
## Anatomy of a provider package
```
providers/<name>/
├── __init__.py # Required. Builds + register()s ProviderSpec
├── config.py # Required. Pydantic Request + Response, both with `provider: Literal["<name>"]`
├── provider.py # Required. TelephonyProvider subclass
├── transport.py # Required. async create_transport(...) -> FastAPIWebsocketTransport
├── serializers.py # Optional but conventional. Re-export from pipecat
├── routes.py # Optional. APIRouter mounted lazily under /api/v1/telephony
└── strategies.py # Optional. Transfer/Hangup strategies for the frame serializer
```
Every file is provider-local. Nothing here imports another provider package.
## The two edits outside this folder
After creating `providers/<name>/`:
1. `providers/__init__.py` — add `<name>` to the import-for-side-effects list. Registration runs at import time.
2. `api/schemas/telephony_config.py` — import `<Name>ConfigurationRequest`/`Response` and add the request to the `TelephonyConfigRequest` `Union[...]` and the response as an optional field on `TelephonyConfigurationResponse`.
If you find yourself editing anything else, re-read the registry plumbing first:
| Want to change... | Source of truth |
| --- | --- |
| Outbound provider lookup | `factory.get_default_telephony_provider`, `get_telephony_provider_by_id`, and `get_telephony_provider_for_run` read `registry.get(name).provider_cls` |
| Stored credentials → constructor dict | `ProviderSpec.config_loader` |
| Audio sample rate / VAD rate | `ProviderSpec.transport_sample_rate` (full `AudioConfig` is built in `pipecat/audio_config.py::create_audio_config`) |
| Which transport runs in `run_pipeline_telephony` | `ProviderSpec.transport_factory` |
| Save-request validation + masked response shape | `ProviderSpec.config_request_cls` / `config_response_cls` |
| Form rendered by the telephony-config UI | `ProviderSpec.ui_metadata` (`ProviderUIField` list) |
| Which credential masks on read | `ui_metadata.fields[*].sensitive=True` (no separate list) |
| Inbound webhook → config row matching | `ProviderSpec.account_id_credential_field` |
| HTTP routes (answer URL, status callbacks) | `providers/<name>/routes.py` (auto-mounted via `importlib`) |
## ProviderSpec — minimum viable shape
```python
SPEC = ProviderSpec(
name="<name>", # registry key, WorkflowRunMode value, stored discriminator
provider_cls=YourProvider,
config_loader=_config_loader, # raw dict from DB → constructor dict
transport_factory=create_transport,
transport_sample_rate=8000, # wire-format rate; pipecat derives the full AudioConfig
config_request_cls=YourProviderConfigurationRequest,
config_response_cls=YourProviderConfigurationResponse,
ui_metadata=ProviderUIMetadata(...), # drives the form UI
account_id_credential_field="api_key", # "" if provider has no account-id concept
)
register(SPEC)
```
`ProviderSpec` is frozen — immutable post-registration. Re-registration with the same instance is a no-op; re-registration with a different instance raises.
## Registration is import-driven, not config-driven
`api/services/telephony/__init__.py` imports `providers/` for side effects. Don't add a registration call elsewhere — by the time `factory`, `audio_config`, or `run_pipeline_telephony` look the spec up, the package init has already executed.
The package init **does not import `routes.py`**`api/routes/telephony.py::_mount_provider_routers()` walks `registry.all_specs()` and uses `importlib.import_module(f"...providers.{spec.name}.routes")`, treating `ModuleNotFoundError` as "no routes for this provider." This is what keeps `from api.services.telephony.base import TelephonyProvider` from fanning out to every route handler in the app. Don't undo it by importing `.routes` from `__init__.py`.
## Conventions
### `provider: Literal["<name>"]` on both Request and Response
Pydantic's discriminated union dispatches on this field. Forgetting `Literal` makes the union accept any provider's payload as yours. Default it to the literal so save calls don't have to send it explicitly.
### Transports load credentials lazily
Always:
```python
from api.services.telephony.factory import load_credentials_for_transport
config = await load_credentials_for_transport(
organization_id, telephony_configuration_id, expected_provider="<name>",
)
```
Never read the org's default config from `transport.py`. The workflow run carries `telephony_configuration_id` in `initial_context` for multi-config orgs; `load_credentials_for_transport` resolves the right row and validates the provider matches.
### `_config_loader` is a pure dict reshape
It runs over `TelephonyConfigurationModel.credentials` (the JSONB column). Don't do I/O in it. Don't pull `from_numbers` from credentials — the factory attaches active phone numbers from `telephony_phone_numbers` after the loader runs, by joining and normalizing addresses.
### Sensitive fields
Mark every credential field `sensitive=True` in `ProviderUIMetadata`. The org routes derive masking from `ui_metadata`, not from a separate hardcoded list. If you re-submit a masked value, `preserve_masked_fields` restores the original — relying on this means you should never write `sensitive=False` on a real secret to "make the form simpler."
### Inbound webhook routing
When multiple configs of the same provider live in one org (e.g. two Twilio sub-accounts), the inbound dispatcher matches the webhook to a config by `credentials[<account_id_credential_field>]`. Set this to whatever your provider stamps on inbound payloads (`account_sid` for Twilio, `auth_id` for Plivo, etc.). Set `""` only when the provider truly has no account-id concept (e.g. ARI — there's at most one config per org).
### `configure_inbound` defaults to no-op
Override only when the provider supports programmatic webhook binding (Plivo `application_id`, Telnyx app config). Markup-response providers that learn the webhook URL from console-side configuration leave the default. Returning `ProviderSyncResult(ok=False, message="...")` surfaces a non-fatal warning to the user without aborting the DB write.
## Reference implementations
Pick the closest shape and copy from it.
| Provider | Pick when... |
| --- | --- |
| `twilio/` | Markup-response (TwiML), HMAC-signed webhooks, conference-style transfers, status callbacks. The most full-featured reference. |
| `plivo/` | Markup-response with multi-callback signature schemes, programmatic answer-URL sync via Application API. |
| `vonage/` | JWT auth, 16 kHz Linear PCM wire format, NCCO JSON responses. |
| `cloudonix/` | SIP-trunk-style with custom transfer/hangup strategies. |
| `telnyx/` | Call-control style — REST calls to answer/stream rather than markup response. |
| `vobiz/` | Body-signed webhooks (signature covers raw bytes). |
| `ari/` | Smallest viable: no `routes.py`, no `verify_inbound_signature`, WebSocket-only, no account-id. |
## What NOT to do
- **Don't import another provider's `provider.py` or `transport.py`.** Cross-provider behavior belongs in `services/telephony/` (e.g. `status_processor`, `ari_manager`, `call_transfer_manager`), not in another provider's package.
- **Don't add a hardcoded provider list anywhere.** If you need to iterate, use `registry.all_specs()` / `registry.names()`.
- **Don't add a route under `routes/telephony.py` for a single provider.** Provider-specific handlers go in `providers/<name>/routes.py`. Cross-provider handlers (`/inbound/run`, `/twiml`) stay in `routes/telephony.py`.
- **Don't import `.routes` from a provider's `__init__.py`.** That's the cycle we deliberately broke — see "Registration is import-driven."
- **Don't write a frontend form for a new provider.** The UI consumes `GET /api/v1/organizations/telephony-providers/metadata` and renders generically from `ProviderUIField`. If a `field.type` you need doesn't exist (`text`/`password`/`textarea`/`string-array`/`number`), extend the renderer in `ui/src/app/(authenticated)/telephony-configurations/` once — not per provider.
- **Don't run a database migration to add a provider.** The discriminator lives in JSONB credentials and a `VARCHAR(64)` `mode` column; nothing in the DB schema knows the set of provider names.
@AGENTS.md

View file

@ -25,7 +25,6 @@ TELNYX_TIMESTAMP_TOLERANCE_SECONDS = 300
TELNYX_PUBLIC_KEY_BYTES = 32
TELNYX_SIGNATURE_BYTES = 64
from api.constants import TELNYX_WEBHOOK_VERIFICATION_OPTIONAL
from api.enums import WorkflowRunMode
from api.services.telephony.base import (
CallInitiationResult,
@ -211,12 +210,6 @@ class TelnyxProvider(TelephonyProvider):
return False
if not self.webhook_public_key:
# REMOVE-AFTER 2026-05-15: transition window. Allow webhooks
# through for configs that haven't added the key yet. Remove this
# branch along with TELNYX_WEBHOOK_VERIFICATION_OPTIONAL after
# the cutoff.
if TELNYX_WEBHOOK_VERIFICATION_OPTIONAL:
return True
logger.error("Missing Telnyx webhook_public_key configuration")
return False

View file

@ -7,7 +7,7 @@ script in `api/services/admin_utils/local_exec.py` is the production
consumer.
"""
from api.services.workflow.node_specs import REGISTRY
from api.services.workflow.node_specs import all_specs
def _build_type_rules() -> tuple[set[str], set[str]]:
@ -16,14 +16,14 @@ def _build_type_rules() -> tuple[set[str], set[str]]:
(max_incoming == 0)."""
src_forbidden: set[str] = set()
tgt_forbidden: set[str] = set()
for name, spec in REGISTRY.items():
for spec in all_specs():
gc = spec.graph_constraints
if gc is None:
continue
if gc.max_outgoing == 0:
src_forbidden.add(name)
src_forbidden.add(spec.name)
if gc.max_incoming == 0:
tgt_forbidden.add(name)
tgt_forbidden.add(spec.name)
return src_forbidden, tgt_forbidden

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,254 @@
"""Single unit that knows the MCP protocol + credentials.
Wraps the vendored Pipecat ``MCPClient`` for connection/session, builds
streamable-HTTP params from a Dograh credential, exposes namespaced
``FunctionSchema``s, and proxies tool calls. Connection failures degrade
(``available = False``) instead of raising the call must survive a
dead MCP server.
"""
from __future__ import annotations
import asyncio
from datetime import timedelta
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set
from loguru import logger
from mcp.client.session_group import StreamableHttpParameters
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.services.mcp_service import MCPClient
from api.services.workflow.tools.mcp_tool import namespace_function_name
from api.utils.credential_auth import build_auth_header
if TYPE_CHECKING:
from api.db.models import ExternalCredentialModel
def build_streamable_http_params(
*,
url: str,
credential: Optional["ExternalCredentialModel"],
timeout_secs: int,
sse_read_timeout_secs: int,
) -> StreamableHttpParameters:
"""Build Pipecat/MCP streamable-HTTP params, injecting the auth header
from an ExternalCredentialModel (reuses the http_api credential path)."""
headers: Optional[Dict[str, str]] = None
if credential is not None:
auth = build_auth_header(credential)
headers = auth or None
return StreamableHttpParameters(
url=url,
headers=headers,
timeout=timedelta(seconds=timeout_secs),
sse_read_timeout=timedelta(seconds=sse_read_timeout_secs),
)
class McpToolSession:
"""One live MCP server connection for the duration of a call."""
def __init__(
self,
*,
tool_uuid: str,
tool_name: str,
url: str,
credential: Optional["ExternalCredentialModel"],
tools_filter: List[str],
timeout_secs: int,
sse_read_timeout_secs: int,
) -> None:
self._tool_uuid = tool_uuid
self._tool_name = tool_name
self._url = url
self._credential = credential
# An empty list is intentionally treated as "no filter (expose all
# tools)" — Pipecat's MCPClient applies a filter only when this is a
# non-empty list, so [] and None are equivalent ("all tools").
self._tools_filter = tools_filter or None
self._timeout_secs = timeout_secs
self._sse_read_timeout_secs = sse_read_timeout_secs
self._client: Optional[MCPClient] = None
self._session: Any = None # mcp.ClientSession (read once after start)
self._schemas: List[FunctionSchema] = []
# namespaced LLM name -> original MCP tool name
self._name_map: Dict[str, str] = {}
self.available: bool = False
async def start(self) -> None:
"""Connect, initialize, and cache the tool list. Never raises —
on any failure the session is marked unavailable."""
try:
params = build_streamable_http_params(
url=self._url,
credential=self._credential,
timeout_secs=self._timeout_secs,
sse_read_timeout_secs=self._sse_read_timeout_secs,
)
self._client = MCPClient(params, tools_filter=self._tools_filter)
await self._client.start()
# Single, isolated touch of Pipecat internals (vendored submodule).
self._session = self._client._active_session
tools_schema = await self._client.get_tools_schema()
fallback = self._tool_uuid[:8] if self._tool_uuid else "server"
for fs in tools_schema.standard_tools:
ns_name = namespace_function_name(
self._tool_name, fs.name, fallback=fallback
)
self._name_map[ns_name] = fs.name
self._schemas.append(
FunctionSchema(
name=ns_name,
description=fs.description,
properties=fs.properties,
required=fs.required,
)
)
self.available = True
logger.info(
f"MCP session ready for tool '{self._tool_name}' "
f"({self._tool_uuid}): {sorted(self._name_map)}"
)
except (KeyboardInterrupt, SystemExit):
raise
except asyncio.CancelledError as e:
# Empirically, a dead/unreachable MCP server does NOT surface as a
# plain Exception here. The real failure is httpx.ConnectError, but
# anyio's streamablehttp_client task group, while tearing down that
# ConnectError, re-surfaces it to our frame as an *internal*
# cancel-scope CancelledError carrying the signature message
# "Cancelled via cancel scope <id>". A genuine *external*
# cancellation (call teardown / shutdown) is a bare CancelledError
# (empty args) or one with an application-chosen message. Type, MRO,
# context chain, and asyncio task.cancelling() are all identical
# between the two, so the anyio scope-signature message is the only
# reliable discriminator. Re-raise genuine external cancellation to
# preserve structured concurrency; degrade only on the anyio
# connect-teardown artifact.
msg = "" if not e.args else str(e.args[0] or "")
if not msg.startswith("Cancelled via cancel scope"):
raise
await self._degrade(e)
except Exception as e: # noqa: BLE001 — see _degrade docstring
# Defensive: if a future Pipecat/httpx version surfaces the connect
# failure directly (e.g. httpx.ConnectError) instead of via the
# anyio cancel-scope artifact above, still degrade gracefully.
await self._degrade(e)
async def _degrade(self, e: BaseException) -> None:
"""Mark this session unavailable and tear down any dangling client so
start() leaves self._client either fully usable or None. The contract
requires graceful degradation on any *connect* failure (never raising
for a dead MCP server) while genuine external cancellation /
KeyboardInterrupt / SystemExit are re-raised by the caller."""
self.available = False
self._schemas = []
self._name_map = {}
# Self-contained cleanup: _client.start() may have succeeded before a
# later step (e.g. get_tools_schema()) failed, leaving an open client.
if self._client is not None:
try:
await self._client.close()
except Exception:
pass
finally:
self._client = None
self._session = None
logger.warning(
f"MCP session unavailable for tool '{self._tool_name}' "
f"({self._tool_uuid}) at {self._url}: {e!r}. "
f"Call proceeds without these tools."
)
@property
def call_timeout_secs(self) -> float:
"""Pipecat function-call timeout for this server's tools. Slightly
longer than the transport read timeout so a slow MCP call surfaces
as a structured tool error (handled in the handler) rather than a
hard pipeline timeout."""
return float(self._sse_read_timeout_secs) + 5.0
def function_schemas(
self, allowed_raw_names: Optional[Set[str]] = None
) -> List[FunctionSchema]:
"""Return cached FunctionSchemas, optionally filtered by raw MCP tool name.
``allowed_raw_names=None`` returns all schemas. An empty set returns none.
Raw names are the pre-namespace MCP tool names (e.g. ``echo``, not
``mcp__slug__echo``).
"""
if allowed_raw_names is None:
return list(self._schemas)
return [
s for s in self._schemas if self._name_map.get(s.name) in allowed_raw_names
]
def discovered_tools(self) -> List[Dict[str, str]]:
"""Raw MCP tool catalog for UI/cache: ``[{name, description}]``
using the *raw* server names (not the namespaced LLM names).
Empty if the session is unavailable."""
out: List[Dict[str, str]] = []
for s in self._schemas:
raw = self._name_map.get(s.name)
if raw is None:
continue
out.append({"name": raw, "description": s.description or ""})
return out
async def call(self, namespaced_name: str, arguments: Dict[str, Any]) -> str:
"""Invoke an MCP tool by its namespaced LLM name. Returns a string
(flattened text content). Raises if the session is unavailable so
the caller can map it to a structured error for the LLM."""
if not self.available or self._session is None:
raise RuntimeError(f"MCP session unavailable for {namespaced_name}")
original = self._name_map.get(namespaced_name)
if original is None:
raise RuntimeError(f"Unknown MCP function {namespaced_name}")
result = await self._session.call_tool(original, arguments=arguments)
text = ""
for content in getattr(result, "content", []) or []:
if getattr(content, "text", None):
text += content.text
return text or "Sorry, the MCP tool returned no content."
async def close(self) -> None:
if self._client is not None:
try:
await self._client.close()
except Exception as e:
logger.warning(f"Error closing MCP session {self._tool_uuid}: {e}")
finally:
self._client = None
self._session = None
async def discover_mcp_tools(
*,
url: str,
credential: Optional["ExternalCredentialModel"],
timeout_secs: int,
sse_read_timeout_secs: int,
) -> List[Dict[str, str]]:
"""Open an ephemeral MCP session, list its tools, close it. Returns
``[{name, description}]`` (raw names). Never raises on any connect
failure returns ``[]``."""
session = McpToolSession(
tool_uuid="discover",
tool_name="discover",
url=url,
credential=credential,
tools_filter=[],
timeout_secs=timeout_secs,
sse_read_timeout_secs=sse_read_timeout_secs,
)
await session.start()
try:
if not session.available:
return []
return session.discovered_tools()
finally:
await session.close()

View file

@ -0,0 +1,19 @@
from __future__ import annotations
from pydantic import BaseModel
from api.services.workflow.node_specs._base import PropertyType
from api.services.workflow.node_specs.model_spec import spec_field
class BaseNodeData(BaseModel):
name: str = spec_field(
...,
min_length=1,
ui_type=PropertyType.string,
display_name="Name",
description="Short identifier shown in the canvas and call logs.",
required=True,
)
is_start: bool = spec_field(default=False, spec_exclude=True)
is_end: bool = spec_field(default=False, spec_exclude=True)

View file

@ -1,10 +1,8 @@
"""Node specification registry.
Adding a new node type:
1. Create a new module under this package, define a `SPEC: NodeSpec`.
2. Add it to the imports + REGISTRY below.
3. The Pydantic discriminated-union variant in dto.py must use the same
`name` value as `SPEC.name`.
Core node specs are generated from the workflow DTO models. Third-party
integration node specs live under `api.services.integrations/<name>/` and
register through the integration registry so they don't need edits here.
"""
from __future__ import annotations
@ -21,8 +19,10 @@ from api.services.workflow.node_specs._base import (
PropertyType,
evaluate_display_options,
)
from api.services.workflow.node_specs.model_spec import build_spec
REGISTRY: dict[str, NodeSpec] = {}
_CORE_SPECS_LOADED = False
def register(spec: NodeSpec) -> NodeSpec:
@ -38,12 +38,23 @@ def register(spec: NodeSpec) -> NodeSpec:
def get_spec(name: str) -> NodeSpec | None:
return REGISTRY.get(name)
_ensure_core_registered()
if name in REGISTRY:
return REGISTRY[name]
from api.services.integrations import get_node_spec
return get_node_spec(name)
def all_specs() -> list[NodeSpec]:
"""All registered specs, sorted by name for stable output."""
return [REGISTRY[name] for name in sorted(REGISTRY)]
_ensure_core_registered()
from api.services.integrations import all_node_specs
specs = {spec.name: spec for spec in REGISTRY.values()}
specs.update({spec.name: spec for spec in all_node_specs()})
return [specs[name] for name in sorted(specs)]
__all__ = [
@ -64,19 +75,15 @@ __all__ = [
]
# Side-effect imports — each module's `register(SPEC)` call populates REGISTRY.
# Keep at module bottom so the registry helpers are defined first.
from api.services.workflow.node_specs import ( # noqa: E402, F401
agent,
end_call,
global_node,
qa,
start_call,
trigger,
webhook,
)
def _ensure_core_registered() -> None:
global _CORE_SPECS_LOADED
if _CORE_SPECS_LOADED:
return
# Wire up registrations from the SPEC constants in each module.
for _module in (start_call, agent, end_call, global_node, trigger, webhook, qa):
register(_module.SPEC)
del _module
from api.services.workflow.dto import _CORE_NODE_DATA_CLASSES
for model_cls in _CORE_NODE_DATA_CLASSES.values():
if model_cls.__node_spec_metadata__.name in REGISTRY:
continue
register(build_spec(model_cls))
_CORE_SPECS_LOADED = True

View file

@ -1,9 +1,9 @@
"""Spec schema for node definitions.
A `NodeSpec` is the single source of truth for a node type. It drives:
- Pydantic validation (the per-type DTOs in dto.py mirror these property types)
- The generic UI renderer (frontend reads specs via /api/v1/node-types)
- The LLM SDK (constructors and JSON-Schema derived from these specs)
`NodeSpec` is the serialized contract exposed to the frontend, MCP tools, and
SDKs. Core workflow node specs are generated from the DTO models plus
model-attached metadata; integration packages may generate them the same way or
register a prebuilt spec object.
Every property's `description` is LLM-readable copy — treat it as production
documentation, not internal notes. Spec lint enforces non-empty descriptions
@ -122,6 +122,16 @@ class PropertyOption(BaseModel):
model_config = ConfigDict(extra="forbid")
def to_mcp_dict(self) -> dict[str, Any]:
"""Lean projection for `get_node_type`: the `value` an LLM writes in
code, plus a `description` when one carries real meaning. The UI
`label` is dropped it's the option's display string, never used
when authoring."""
out: dict[str, Any] = {"value": self.value}
if self.description:
out["description"] = self.description
return out
class PropertySpec(BaseModel):
"""Single field on a node.
@ -175,6 +185,43 @@ class PropertySpec(BaseModel):
model_config = ConfigDict(extra="forbid")
def to_mcp_dict(self) -> dict[str, Any]:
"""Lean projection of this property for the `get_node_type` MCP tool.
Keeps only what an LLM needs to author a valid value: name, type,
description, llm_hint, requiredness, default, enum options, nested
row properties, and validation bounds. UI-rendering concerns
(`display_name`, `placeholder`, `display_options`, `editor`,
`extra`) and null/empty fields are omitted they're noise in the
model's context and never appear in authored SDK code.
"""
out: dict[str, Any] = {
"name": self.name,
"type": self.type.value,
"description": self.description,
}
if self.llm_hint:
out["llm_hint"] = self.llm_hint
if self.required:
out["required"] = True
if self.default is not None:
out["default"] = self.default
if self.options:
out["options"] = [opt.to_mcp_dict() for opt in self.options]
if self.properties:
out["properties"] = [prop.to_mcp_dict() for prop in self.properties]
for constraint in (
"min_value",
"max_value",
"min_length",
"max_length",
"pattern",
):
value = getattr(self, constraint)
if value is not None:
out[constraint] = value
return out
PropertySpec.model_rebuild()
@ -222,3 +269,33 @@ class NodeSpec(BaseModel):
graph_constraints: Optional[GraphConstraints] = None
model_config = ConfigDict(extra="forbid")
def to_mcp_dict(self) -> dict[str, Any]:
"""Lean projection of this spec for the `get_node_type` MCP tool.
Drops node-level UI metadata (`display_name`, `category`, `icon`,
`version`) and the per-property rendering concerns trimmed by
`PropertySpec.to_mcp_dict`, leaving just the authoring-relevant
schema the LLM consumes when composing a workflow. The full spec is
still served verbatim to the frontend renderer (REST `node-types`
route) and the SDK codegen / TS validator (`ts_bridge`), which need
the dropped fields.
"""
out: dict[str, Any] = {
"name": self.name,
"description": self.description,
}
if self.llm_hint:
out["llm_hint"] = self.llm_hint
out["properties"] = [prop.to_mcp_dict() for prop in self.properties]
if self.examples:
out["examples"] = [
ex.model_dump(mode="json", exclude_none=True) for ex in self.examples
]
if self.graph_constraints:
constraints = self.graph_constraints.model_dump(
mode="json", exclude_none=True
)
if constraints:
out["graph_constraints"] = constraints
return out

View file

@ -1,168 +0,0 @@
"""Spec for the Agent node — the workhorse mid-call node where the LLM
executes a focused conversational step with optional tools and documents."""
from api.services.workflow.node_specs._base import (
DisplayOptions,
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertyOption,
PropertySpec,
PropertyType,
)
SPEC = NodeSpec(
name="agentNode",
display_name="Agent Node",
description="Conversational step — the LLM runs one focused exchange.",
llm_hint=(
"Mid-call step executed by the LLM. Most workflows are a chain of "
"agent nodes connected by edges that describe transition conditions. "
"Each agent node can invoke tools and reference documents."
),
category=NodeCategory.call_node,
icon="Headset",
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Name",
description=(
"Short identifier for this step (e.g., 'Qualify Budget'). "
"Appears in call logs and edge transition tools."
),
required=True,
min_length=1,
default="Agent",
),
PropertySpec(
name="prompt",
type=PropertyType.mention_textarea,
display_name="Prompt",
description=(
"Agent system prompt for this step. Supports "
"{{template_variables}} from extraction or pre-call fetch."
),
required=True,
min_length=1,
placeholder="Ask the caller about their budget and timeline.",
),
PropertySpec(
name="allow_interrupt",
type=PropertyType.boolean,
display_name="Allow Interruption",
description=(
"When true, the user can interrupt the agent mid-utterance. "
"Set false for non-interruptible disclosures."
),
default=True,
),
PropertySpec(
name="add_global_prompt",
type=PropertyType.boolean,
display_name="Add Global Prompt",
description=(
"When true and a Global node exists, prepends the global "
"prompt to this node's prompt at runtime."
),
default=True,
),
PropertySpec(
name="extraction_enabled",
type=PropertyType.boolean,
display_name="Enable Variable Extraction",
description=(
"When true, runs an LLM extraction pass on transition out of "
"this node to capture variables from the conversation."
),
default=False,
),
PropertySpec(
name="extraction_prompt",
type=PropertyType.string,
display_name="Extraction Prompt",
description="Overall instructions guiding variable extraction.",
display_options=DisplayOptions(show={"extraction_enabled": [True]}),
editor="textarea",
),
PropertySpec(
name="extraction_variables",
type=PropertyType.fixed_collection,
display_name="Variables to Extract",
description=(
"Each entry declares one variable to capture from the "
"conversation, with its name, type, and per-variable hint."
),
display_options=DisplayOptions(show={"extraction_enabled": [True]}),
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Variable Name",
description="snake_case identifier used downstream.",
required=True,
),
PropertySpec(
name="type",
type=PropertyType.options,
display_name="Type",
description="Data type of the extracted value.",
required=True,
default="string",
options=[
PropertyOption(value="string", label="String"),
PropertyOption(value="number", label="Number"),
PropertyOption(value="boolean", label="Boolean"),
],
),
PropertySpec(
name="prompt",
type=PropertyType.string,
display_name="Extraction Hint",
description="Per-variable hint describing what to look for.",
editor="textarea",
),
],
),
PropertySpec(
name="tool_uuids",
type=PropertyType.tool_refs,
display_name="Tools",
description="Tools the agent can invoke during this step.",
llm_hint="List of tool UUIDs from `list_tools`.",
),
PropertySpec(
name="document_uuids",
type=PropertyType.document_refs,
display_name="Knowledge Base Documents",
description="Documents the agent can reference during this step.",
llm_hint="List of document UUIDs from `list_documents`.",
),
],
examples=[
NodeExample(
name="qualify_lead",
data={
"name": "Qualify Budget",
"prompt": "Ask about budget and timeline. Capture both before transitioning.",
"allow_interrupt": True,
"extraction_enabled": True,
"extraction_prompt": "Extract budget amount and rough timeline.",
"extraction_variables": [
{
"name": "budget_usd",
"type": "number",
"prompt": "Stated budget in USD",
},
{
"name": "timeline",
"type": "string",
"prompt": "When they want to start",
},
],
},
),
],
graph_constraints=GraphConstraints(min_incoming=1),
)

View file

@ -0,0 +1,44 @@
DEFAULT_QA_SYSTEM_PROMPT = """You are a QA analyst evaluating a specific segment of a voice AI conversation.
## Node Purpose
{{node_summary}}
## Previous Conversation Context (For start of conversation, previous conversation summary can be empty.)
{{previous_conversation_summary}}
## Tags to evaluate
Examine the conversation carefully and identify which of the following tags apply:
- UNCLEAR_CONVERSATION - The conversation is not coherent or clear, messages don't connect logically
- ASSISTANT_IN_LOOP - The assistant asks the same question multiple times or gets stuck repeating itself
- ASSISTANT_REPLY_IMPROPER - The assistant did not reply properly to the user's question/query or seems confused by what the user said
- USER_FRUSTRATED - The user seems angry, frustrated, or is complaining about something in the call
- USER_NOT_UNDERSTANDING - The user explicitly says they don't understand or repeatedly asks for clarification
- HEARING_ISSUES - Either party can't hear the other ("hello?", "are you there?", "can you hear me?")
- DEAD_AIR - Unusually long silences in the conversation (use the timestamps to judge)
- USER_REQUESTING_FEATURE - The user asks for something the assistant can't fulfill
- ASSISTANT_LACKS_EMPATHY - The assistant ignores the user's personal situation or emotional state and continues pitching or pushing the agenda.
- USER_DETECTS_AI - The user suspects or identifies that they are talking to an AI/robot/bot rather than a real human.
## Call metrics (pre-computed)
Use these alongside the transcript for your analysis:
{{metrics}}
## Output format
Return ONLY a valid JSON object (no markdown):
{
"tags": [
{
"tag": "TAG_NAME",
"reason": "Short reason with evidence from the transcript"
}
],
"overall_sentiment": "positive|neutral|negative",
"call_quality_score": <1-10>,
"summary": "1-2 sentence summary of this segment"
}
If no tags apply, return an empty tags list. Always provide sentiment, score, and summary."""

View file

@ -1,141 +0,0 @@
"""Spec for the End Call node — terminal node that wraps up a conversation
and optionally extracts variables before hangup."""
from api.services.workflow.node_specs._base import (
DisplayOptions,
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertyOption,
PropertySpec,
PropertyType,
)
SPEC = NodeSpec(
name="endCall",
display_name="End Call",
description="Closes the conversation and hangs up.",
llm_hint=(
"Terminal node that politely closes the conversation. Variable "
"extraction can run before hangup. A workflow can have multiple "
"endCall nodes reached via different edge conditions."
),
category=NodeCategory.call_node,
icon="OctagonX",
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Name",
description=(
"Short identifier shown in call logs. Should describe the "
"ending context (e.g., 'Successful close', 'Polite decline')."
),
required=True,
min_length=1,
default="End Call",
),
PropertySpec(
name="prompt",
type=PropertyType.mention_textarea,
display_name="Prompt",
description=(
"Agent system prompt for the closing exchange. Supports "
"{{template_variables}} from extraction or pre-call fetch."
),
required=True,
min_length=1,
placeholder="Thank the caller and confirm next steps before ending the call.",
),
PropertySpec(
name="add_global_prompt",
type=PropertyType.boolean,
display_name="Add Global Prompt",
description=(
"When true and a Global node exists, prepends the global "
"prompt to this node's prompt at runtime."
),
default=False,
),
PropertySpec(
name="extraction_enabled",
type=PropertyType.boolean,
display_name="Enable Variable Extraction",
description=(
"When true, runs an LLM extraction pass before hangup to "
"capture variables from the conversation."
),
default=False,
),
PropertySpec(
name="extraction_prompt",
type=PropertyType.string,
display_name="Extraction Prompt",
description=(
"Overall instructions guiding how variables should be "
"extracted from the conversation."
),
display_options=DisplayOptions(show={"extraction_enabled": [True]}),
editor="textarea",
),
PropertySpec(
name="extraction_variables",
type=PropertyType.fixed_collection,
display_name="Variables to Extract",
description=(
"Each entry declares one variable to capture from the "
"conversation, with its name, data type, and a per-variable "
"extraction hint."
),
display_options=DisplayOptions(show={"extraction_enabled": [True]}),
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Variable Name",
description="snake_case identifier used downstream.",
required=True,
),
PropertySpec(
name="type",
type=PropertyType.options,
display_name="Type",
description="The data type of the extracted value.",
required=True,
default="string",
options=[
PropertyOption(value="string", label="String"),
PropertyOption(value="number", label="Number"),
PropertyOption(value="boolean", label="Boolean"),
],
),
PropertySpec(
name="prompt",
type=PropertyType.string,
display_name="Extraction Hint",
description=(
"Per-variable hint describing what to look for in "
"the conversation."
),
editor="textarea",
),
],
),
],
examples=[
NodeExample(
name="successful_close",
data={
"name": "Successful Close",
"prompt": "Confirm the appointment time, thank the caller, and end the call.",
"add_global_prompt": False,
},
),
],
graph_constraints=GraphConstraints(
min_incoming=1,
min_outgoing=0,
max_outgoing=0,
),
)

View file

@ -1,77 +0,0 @@
"""Spec for the Global node — system-level instructions appended to every
agent node that opts in via `add_global_prompt`."""
from api.services.workflow.node_specs._base import (
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertySpec,
PropertyType,
)
SPEC = NodeSpec(
name="globalNode",
display_name="Global Node",
description="Persona/tone appended to every agent node's prompt.",
llm_hint=(
"System-level prompt appended to every prompted node whose "
"`add_global_prompt` is true. Use it for persona, tone, and shared "
"rules that apply across the entire conversation. At most one "
"global node per workflow."
),
category=NodeCategory.global_node,
icon="Globe",
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Name",
description=(
"Short identifier shown in the canvas and call logs. Has no "
"runtime effect."
),
required=True,
min_length=1,
default="Global Node",
),
PropertySpec(
name="prompt",
type=PropertyType.mention_textarea,
display_name="Global Prompt",
description=(
"Text appended to every prompted node's system prompt when "
"that node has `add_global_prompt=true`. Supports "
"{{template_variables}}."
),
required=True,
min_length=1,
placeholder="You are a friendly assistant calling on behalf of {{company_name}}.",
default=(
"You are a helpful assistant whose mode of interaction with "
"the user is voice. So don't use any special characters which "
"can not be pronounced. Use short sentences and simple language."
),
),
],
examples=[
NodeExample(
name="basic_persona",
description="Establishes a consistent persona across the call.",
data={
"name": "Persona",
"prompt": (
"You are Sarah, a polite and warm representative from "
"Acme Corp. Always thank the caller for their time and "
"speak in short conversational sentences."
),
},
),
],
graph_constraints=GraphConstraints(
min_incoming=0,
max_incoming=0,
min_outgoing=0,
max_outgoing=0,
),
)

View file

@ -0,0 +1,404 @@
from __future__ import annotations
from dataclasses import dataclass
from dataclasses import field as dataclass_field
from enum import Enum
from types import NoneType
from typing import Any, Callable, Literal, get_args, get_origin
from pydantic import BaseModel, Field
from pydantic.fields import FieldInfo, PydanticUndefined
from api.services.workflow.node_specs._base import (
DisplayOptions,
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertyOption,
PropertySpec,
PropertyType,
)
_SPEC_FIELD_META_KEY = "__dograh_spec_field__"
_UNSET = object()
@dataclass(frozen=True)
class NodeSpecMetadata:
name: str
display_name: str
description: str
category: NodeCategory
icon: str
llm_hint: str | None = None
version: str = "1.0.0"
examples: tuple[NodeExample, ...] = ()
graph_constraints: GraphConstraints | None = None
property_order: tuple[str, ...] = ()
field_overrides: dict[str, dict[str, Any]] = dataclass_field(default_factory=dict)
def spec_field(
*field_args: Any,
ui_type: PropertyType | str | None = None,
display_name: str | None = None,
llm_hint: str | None = None,
required: bool | None = None,
spec_default: Any = _UNSET,
placeholder: str | None = None,
display_options: DisplayOptions | None = None,
options: list[PropertyOption] | None = None,
editor: str | None = None,
extra: dict[str, Any] | None = None,
spec_exclude: bool = False,
min_value: float | None = None,
max_value: float | None = None,
min_length: int | None = None,
max_length: int | None = None,
pattern: str | None = None,
**field_kwargs: Any,
):
json_schema_extra = dict(field_kwargs.pop("json_schema_extra", {}) or {})
json_schema_extra[_SPEC_FIELD_META_KEY] = {
"ui_type": ui_type.value if isinstance(ui_type, PropertyType) else ui_type,
"display_name": display_name,
"llm_hint": llm_hint,
"required": required,
"placeholder": placeholder,
"display_options": display_options,
"options": options,
"editor": editor,
"extra": extra or {},
"spec_exclude": spec_exclude,
"min_value": min_value,
"max_value": max_value,
"min_length": min_length,
"max_length": max_length,
"pattern": pattern,
}
if spec_default is not _UNSET:
json_schema_extra[_SPEC_FIELD_META_KEY]["spec_default"] = spec_default
return Field(*field_args, json_schema_extra=json_schema_extra, **field_kwargs)
def node_spec(
*,
name: str,
display_name: str,
description: str,
category: NodeCategory,
icon: str,
llm_hint: str | None = None,
version: str = "1.0.0",
examples: list[NodeExample] | tuple[NodeExample, ...] = (),
graph_constraints: GraphConstraints | None = None,
property_order: list[str] | tuple[str, ...] = (),
field_overrides: dict[str, dict[str, Any]] | None = None,
) -> Callable[[type[BaseModel]], type[BaseModel]]:
metadata = NodeSpecMetadata(
name=name,
display_name=display_name,
description=description,
category=category,
icon=icon,
llm_hint=llm_hint,
version=version,
examples=tuple(examples),
graph_constraints=graph_constraints,
property_order=tuple(property_order),
field_overrides=field_overrides or {},
)
def decorator(model_cls: type[BaseModel]) -> type[BaseModel]:
setattr(model_cls, "__node_spec_metadata__", metadata)
return model_cls
return decorator
def build_spec(model_cls: type[BaseModel]) -> NodeSpec:
metadata: NodeSpecMetadata | None = getattr(
model_cls, "__node_spec_metadata__", None
)
if metadata is None:
raise ValueError(f"{model_cls.__name__} is missing __node_spec_metadata__")
properties: list[PropertySpec] = []
for name, field in model_cls.model_fields.items():
prop = _build_property_spec(model_cls, name, field)
if prop is not None:
properties.append(prop)
properties = _sort_properties(metadata.name, properties, metadata.property_order)
return NodeSpec(
name=metadata.name,
display_name=metadata.display_name,
description=metadata.description,
llm_hint=metadata.llm_hint,
category=metadata.category,
icon=metadata.icon,
version=metadata.version,
properties=properties,
examples=list(metadata.examples),
graph_constraints=metadata.graph_constraints,
)
def _sort_properties(
spec_name: str,
properties: list[PropertySpec],
property_order: tuple[str, ...],
) -> list[PropertySpec]:
if not property_order:
return properties
property_names = {prop.name for prop in properties}
missing = [name for name in property_order if name not in property_names]
if missing:
raise ValueError(
f"{spec_name}: property_order references unknown properties: {missing}"
)
order_map = {name: idx for idx, name in enumerate(property_order)}
ordered = sorted(
enumerate(properties),
key=lambda item: (order_map.get(item[1].name, len(order_map)), item[0]),
)
return [prop for _, prop in ordered]
def _build_property_spec(
owner_cls: type[BaseModel],
field_name: str,
field: FieldInfo,
) -> PropertySpec | None:
meta = _merged_field_meta(owner_cls, field_name, field)
if meta.get("spec_exclude"):
return None
prop_type = _resolve_property_type(field.annotation, meta)
nested_properties = _resolve_nested_properties(field.annotation, prop_type)
options = _resolve_options(field.annotation, meta, prop_type)
min_value, max_value, min_length, max_length, pattern = _resolve_constraints(
field, meta
)
description = meta.get("description") or field.description
if not description:
raise ValueError(f"{owner_cls.__name__}.{field_name} is missing a description")
return PropertySpec(
name=field_name,
type=prop_type,
display_name=meta.get("display_name") or _humanize_identifier(field_name),
description=description,
llm_hint=meta.get("llm_hint"),
default=_resolve_default(field, meta),
required=_resolve_required(field, meta),
placeholder=meta.get("placeholder"),
display_options=meta.get("display_options"),
options=options,
properties=nested_properties,
min_value=min_value,
max_value=max_value,
min_length=min_length,
max_length=max_length,
pattern=pattern,
editor=meta.get("editor"),
extra=meta.get("extra") or {},
)
def _merged_field_meta(
owner_cls: type[BaseModel],
field_name: str,
field: FieldInfo,
) -> dict[str, Any]:
field_meta = {}
if isinstance(field.json_schema_extra, dict):
field_meta = dict(field.json_schema_extra.get(_SPEC_FIELD_META_KEY, {}) or {})
metadata: NodeSpecMetadata | None = getattr(
owner_cls, "__node_spec_metadata__", None
)
override = (
dict(metadata.field_overrides.get(field_name, {}) or {})
if metadata is not None
else {}
)
merged = dict(field_meta)
merged.update(override)
return merged
def _resolve_property_type(annotation: Any, meta: dict[str, Any]) -> PropertyType:
ui_type = meta.get("ui_type")
if ui_type:
return PropertyType(ui_type)
inner = _strip_optional(annotation)
origin = get_origin(inner)
args = get_args(inner)
if origin is list:
item_type = _strip_optional(args[0]) if args else Any
if isinstance(item_type, type) and issubclass(item_type, BaseModel):
return PropertyType.fixed_collection
raise ValueError(
"List-valued fields must declare an explicit ui_type unless they wrap a "
f"BaseModel row type (field annotation: {annotation!r})."
)
if _is_enum(inner) or _is_literal(inner):
return PropertyType.options
if inner in (str,):
return PropertyType.string
if inner in (int, float):
return PropertyType.number
if inner is bool:
return PropertyType.boolean
if inner in (dict, Any) or origin is dict:
return PropertyType.json
raise ValueError(f"Unable to derive PropertyType for annotation {annotation!r}")
def _resolve_nested_properties(
annotation: Any,
prop_type: PropertyType,
) -> list[PropertySpec] | None:
if prop_type != PropertyType.fixed_collection:
return None
inner = _strip_optional(annotation)
args = get_args(inner)
if not args:
raise ValueError(
f"fixed_collection field annotation is missing row type: {annotation!r}"
)
row_type = _strip_optional(args[0])
if not isinstance(row_type, type) or not issubclass(row_type, BaseModel):
raise ValueError(
f"fixed_collection rows must be BaseModel subclasses: {annotation!r}"
)
properties: list[PropertySpec] = []
for field_name, field in row_type.model_fields.items():
prop = _build_property_spec(row_type, field_name, field)
if prop is not None:
properties.append(prop)
return properties
def _resolve_options(
annotation: Any,
meta: dict[str, Any],
prop_type: PropertyType,
) -> list[PropertyOption] | None:
if prop_type not in (PropertyType.options, PropertyType.multi_options):
return meta.get("options")
if meta.get("options"):
return meta["options"]
inner = _strip_optional(annotation)
if prop_type == PropertyType.multi_options:
inner = _strip_optional(get_args(inner)[0])
if _is_enum(inner):
return [
PropertyOption(
value=member.value, label=_humanize_option_label(member.value)
)
for member in inner
]
if _is_literal(inner):
return [
PropertyOption(value=value, label=_humanize_option_label(value))
for value in get_args(inner)
if value is not None
]
return None
def _resolve_constraints(
field: FieldInfo,
meta: dict[str, Any],
) -> tuple[float | None, float | None, int | None, int | None, str | None]:
min_value = meta.get("min_value")
max_value = meta.get("max_value")
min_length = meta.get("min_length")
max_length = meta.get("max_length")
pattern = meta.get("pattern")
for item in field.metadata:
if min_value is None:
if hasattr(item, "ge") and item.ge is not None:
min_value = item.ge
elif hasattr(item, "gt") and item.gt is not None:
min_value = item.gt
if max_value is None:
if hasattr(item, "le") and item.le is not None:
max_value = item.le
elif hasattr(item, "lt") and item.lt is not None:
max_value = item.lt
if (
min_length is None
and hasattr(item, "min_length")
and item.min_length is not None
):
min_length = item.min_length
if (
max_length is None
and hasattr(item, "max_length")
and item.max_length is not None
):
max_length = item.max_length
if pattern is None and hasattr(item, "pattern") and item.pattern is not None:
pattern = item.pattern
return min_value, max_value, min_length, max_length, pattern
def _resolve_default(field: FieldInfo, meta: dict[str, Any]) -> Any:
if "spec_default" in meta:
return meta["spec_default"]
if field.default is not PydanticUndefined:
return field.default
return None
def _resolve_required(field: FieldInfo, meta: dict[str, Any]) -> bool:
if meta.get("required") is not None:
return bool(meta["required"])
return bool(field.is_required())
def _strip_optional(annotation: Any) -> Any:
origin = get_origin(annotation)
if origin is None:
return annotation
args = [arg for arg in get_args(annotation) if arg is not NoneType]
if len(args) == 1 and len(args) != len(get_args(annotation)):
return args[0]
return annotation
def _is_enum(annotation: Any) -> bool:
return isinstance(annotation, type) and issubclass(annotation, Enum)
def _is_literal(annotation: Any) -> bool:
return get_origin(annotation) is Literal
def _humanize_identifier(name: str) -> str:
return name.replace("_", " ").strip().title()
def _humanize_option_label(value: Any) -> str:
if isinstance(value, str):
return value.replace("_", " ").replace("-", " ").strip().title()
return str(value)

View file

@ -1,203 +0,0 @@
"""Spec for the QA Analysis node — runs an LLM quality review on the call
transcript after completion."""
from api.services.workflow.node_specs._base import (
DisplayOptions,
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertyOption,
PropertySpec,
PropertyType,
)
DEFAULT_QA_SYSTEM_PROMPT = """You are a QA analyst evaluating a specific segment of a voice AI conversation.
## Node Purpose
{{node_summary}}
## Previous Conversation Context (For start of conversation, previous conversation summary can be empty.)
{{previous_conversation_summary}}
## Tags to evaluate
Examine the conversation carefully and identify which of the following tags apply:
- UNCLEAR_CONVERSATION - The conversation is not coherent or clear, messages don't connect logically
- ASSISTANT_IN_LOOP - The assistant asks the same question multiple times or gets stuck repeating itself
- ASSISTANT_REPLY_IMPROPER - The assistant did not reply properly to the user's question/query or seems confused by what the user said
- USER_FRUSTRATED - The user seems angry, frustrated, or is complaining about something in the call
- USER_NOT_UNDERSTANDING - The user explicitly says they don't understand or repeatedly asks for clarification
- HEARING_ISSUES - Either party can't hear the other ("hello?", "are you there?", "can you hear me?")
- DEAD_AIR - Unusually long silences in the conversation (use the timestamps to judge)
- USER_REQUESTING_FEATURE - The user asks for something the assistant can't fulfill
- ASSISTANT_LACKS_EMPATHY - The assistant ignores the user's personal situation or emotional state and continues pitching or pushing the agenda.
- USER_DETECTS_AI - The user suspects or identifies that they are talking to an AI/robot/bot rather than a real human.
## Call metrics (pre-computed)
Use these alongside the transcript for your analysis:
{{metrics}}
## Output format
Return ONLY a valid JSON object (no markdown):
{
"tags": [
{
"tag": "TAG_NAME",
"reason": "Short reason with evidence from the transcript"
}
],
"overall_sentiment": "positive|neutral|negative",
"call_quality_score": <1-10>,
"summary": "1-2 sentence summary of this segment"
}
If no tags apply, return an empty tags list. Always provide sentiment, score, and summary."""
SPEC = NodeSpec(
name="qa",
display_name="QA Analysis",
description="Run LLM quality analysis on the call transcript.",
llm_hint=(
"Runs an LLM quality review on the call transcript after completion. "
"Per-node analysis splits the conversation by node and evaluates each "
"segment against the configured system prompt. Sampling, minimum "
"duration, and voicemail filters are supported."
),
category=NodeCategory.integration,
icon="ClipboardCheck",
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Name",
description="Short identifier for this QA configuration.",
required=True,
min_length=1,
default="QA Analysis",
),
PropertySpec(
name="qa_enabled",
type=PropertyType.boolean,
display_name="Enabled",
description="When false, the QA run is skipped.",
default=True,
),
PropertySpec(
name="qa_system_prompt",
type=PropertyType.string,
display_name="System Prompt",
description=(
"Instructions to the QA reviewer LLM. Supports placeholders: "
"`{node_summary}`, `{previous_conversation_summary}`, "
"`{transcript}`, `{metrics}`."
),
editor="textarea",
default=DEFAULT_QA_SYSTEM_PROMPT,
),
PropertySpec(
name="qa_min_call_duration",
type=PropertyType.number,
display_name="Minimum Call Duration (seconds)",
description="Calls shorter than this are skipped.",
default=15,
min_value=0,
),
PropertySpec(
name="qa_voicemail_calls",
type=PropertyType.boolean,
display_name="Include Voicemail Calls",
description="When false, calls flagged as voicemail are skipped.",
default=False,
),
PropertySpec(
name="qa_sample_rate",
type=PropertyType.number,
display_name="Sample Rate (%)",
description=(
"Percent of eligible calls QA'd. 100 means every call; lower "
"values use random sampling."
),
default=100,
min_value=1,
max_value=100,
),
# ---- LLM configuration ----
PropertySpec(
name="qa_use_workflow_llm",
type=PropertyType.boolean,
display_name="Use Workflow's LLM",
description=(
"When true, the QA pass uses the same LLM the workflow runs "
"with. Set false to specify a separate provider/model."
),
default=True,
),
PropertySpec(
name="qa_provider",
type=PropertyType.options,
display_name="QA LLM Provider",
description="LLM provider used for the QA pass.",
display_options=DisplayOptions(show={"qa_use_workflow_llm": [False]}),
options=[
PropertyOption(value="openai", label="OpenAI"),
PropertyOption(value="azure", label="Azure OpenAI"),
PropertyOption(value="openrouter", label="OpenRouter"),
PropertyOption(value="anthropic", label="Anthropic"),
],
),
PropertySpec(
name="qa_model",
type=PropertyType.string,
display_name="QA Model",
description=(
"Model identifier (e.g., 'gpt-4o', 'claude-sonnet-4-6'). "
"Provider-specific."
),
display_options=DisplayOptions(show={"qa_use_workflow_llm": [False]}),
default="default",
),
PropertySpec(
name="qa_api_key",
type=PropertyType.string,
display_name="API Key",
description="API key for the chosen provider.",
display_options=DisplayOptions(show={"qa_use_workflow_llm": [False]}),
),
PropertySpec(
name="qa_endpoint",
type=PropertyType.url,
display_name="Azure Endpoint",
description="Required for the Azure provider.",
display_options=DisplayOptions(
show={"qa_use_workflow_llm": [False], "qa_provider": ["azure"]}
),
),
],
examples=[
NodeExample(
name="basic_qa",
data={
"name": "Compliance Check",
"qa_enabled": True,
"qa_system_prompt": (
"You are a compliance reviewer. Review the transcript and "
"produce a JSON object with `tags`, `summary`, "
"`call_quality_score`, and `overall_sentiment`."
),
"qa_min_call_duration": 30,
"qa_sample_rate": 100,
},
),
],
# QA runs post-call against the saved transcript (run_integrations
# scans by type), never as a graph step. Reject any edge into or out
# of a QA node.
graph_constraints=GraphConstraints(
min_incoming=0, max_incoming=0, min_outgoing=0, max_outgoing=0
),
)

View file

@ -1,250 +0,0 @@
"""Spec for the Start Call node — the single entry point of every workflow.
Carries greeting, pre-call data fetch, and the same prompt/extraction/tools
fields as agent nodes."""
from api.services.workflow.node_specs._base import (
DisplayOptions,
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertyOption,
PropertySpec,
PropertyType,
)
SPEC = NodeSpec(
name="startCall",
display_name="Start Call",
description="Entry point of the workflow — plays a greeting and opens the conversation.",
llm_hint=(
"The entry point of every workflow (exactly one required). Plays an "
"optional greeting, can fetch context from an external API before "
"the call begins, and executes the first conversational turn."
),
category=NodeCategory.call_node,
icon="Play",
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Name",
description="Short identifier shown in the canvas and call logs.",
required=True,
min_length=1,
default="Start Call",
),
# ---- Greeting (variant via greeting_type) ----
PropertySpec(
name="greeting_type",
type=PropertyType.options,
display_name="Greeting Type",
description=(
"Whether the optional greeting is spoken via TTS from text "
"or played from a pre-recorded audio file."
),
default="text",
options=[
PropertyOption(value="text", label="Text (TTS)"),
PropertyOption(value="audio", label="Pre-recorded Audio"),
],
),
PropertySpec(
name="greeting",
type=PropertyType.string,
display_name="Greeting Text",
description=(
"Text spoken via TTS at the start of the call. Supports "
"{{template_variables}}. Leave empty to skip the greeting."
),
display_options=DisplayOptions(show={"greeting_type": ["text"]}),
editor="textarea",
placeholder="Hi {{first_name}}, this is Sarah from Acme.",
),
PropertySpec(
name="greeting_recording_id",
type=PropertyType.recording_ref,
display_name="Greeting Recording",
description="Pre-recorded audio file played at the start of the call.",
llm_hint=(
"Value is the `recording_id` string. Use the `list_recordings` "
"MCP tool to discover available recordings."
),
display_options=DisplayOptions(show={"greeting_type": ["audio"]}),
),
PropertySpec(
name="prompt",
type=PropertyType.mention_textarea,
display_name="Prompt",
description=(
"Agent system prompt for the opening turn. Supports "
"{{template_variables}} from pre-call fetch and the initial context."
),
required=True,
min_length=1,
placeholder="Greet the caller warmly and ask how you can help today.",
),
# ---- Behavior toggles ----
PropertySpec(
name="allow_interrupt",
type=PropertyType.boolean,
display_name="Allow Interruption",
description=("When true, the user can interrupt the agent mid-utterance."),
default=False,
),
PropertySpec(
name="add_global_prompt",
type=PropertyType.boolean,
display_name="Add Global Prompt",
description=(
"When true and a Global node exists, prepends the global "
"prompt to this node's prompt at runtime."
),
default=True,
),
PropertySpec(
name="delayed_start",
type=PropertyType.boolean,
display_name="Delayed Start",
description=(
"When true, the agent waits before speaking after pickup. "
"Useful for outbound calls where the called party needs a "
"moment to settle."
),
default=False,
),
PropertySpec(
name="delayed_start_duration",
type=PropertyType.number,
display_name="Delay Duration (seconds)",
description="Seconds to wait before the agent speaks. 0.110.",
default=2.0,
min_value=0.1,
max_value=10.0,
display_options=DisplayOptions(show={"delayed_start": [True]}),
),
# ---- Variable extraction ----
PropertySpec(
name="extraction_enabled",
type=PropertyType.boolean,
display_name="Enable Variable Extraction",
description=(
"When true, runs an LLM extraction pass on transition out of "
"this node to capture variables from the opening turn."
),
default=False,
),
PropertySpec(
name="extraction_prompt",
type=PropertyType.string,
display_name="Extraction Prompt",
description="Overall instructions guiding variable extraction.",
display_options=DisplayOptions(show={"extraction_enabled": [True]}),
editor="textarea",
),
PropertySpec(
name="extraction_variables",
type=PropertyType.fixed_collection,
display_name="Variables to Extract",
description=(
"Each entry declares one variable to capture, with its name, "
"data type, and per-variable extraction hint."
),
display_options=DisplayOptions(show={"extraction_enabled": [True]}),
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Variable Name",
description="snake_case identifier used downstream.",
required=True,
),
PropertySpec(
name="type",
type=PropertyType.options,
display_name="Type",
description="Data type of the extracted value.",
required=True,
default="string",
options=[
PropertyOption(value="string", label="String"),
PropertyOption(value="number", label="Number"),
PropertyOption(value="boolean", label="Boolean"),
],
),
PropertySpec(
name="prompt",
type=PropertyType.string,
display_name="Extraction Hint",
description="Per-variable hint describing what to look for.",
editor="textarea",
),
],
),
# ---- Tools / documents ----
PropertySpec(
name="tool_uuids",
type=PropertyType.tool_refs,
display_name="Tools",
description="Tools the agent can invoke during the opening turn.",
llm_hint="List of tool UUIDs from `list_tools`.",
),
PropertySpec(
name="document_uuids",
type=PropertyType.document_refs,
display_name="Knowledge Base Documents",
description="Documents the agent can reference.",
llm_hint="List of document UUIDs from `list_documents`.",
),
# ---- Pre-call data fetch (advanced) ----
PropertySpec(
name="pre_call_fetch_enabled",
type=PropertyType.boolean,
display_name="Pre-Call Data Fetch",
description=(
"When true, makes a POST request to an external API before "
"the call starts and merges the JSON response into the call "
"context as template variables."
),
default=False,
),
PropertySpec(
name="pre_call_fetch_url",
type=PropertyType.url,
display_name="Endpoint URL",
description=(
"URL the pre-call POST request is sent to. The request body "
"includes caller and called numbers."
),
display_options=DisplayOptions(show={"pre_call_fetch_enabled": [True]}),
placeholder="https://api.example.com/customer-lookup",
),
PropertySpec(
name="pre_call_fetch_credential_uuid",
type=PropertyType.credential_ref,
display_name="Authentication",
description="Optional credential attached to the pre-call request.",
llm_hint="Credential UUID from `list_credentials`.",
display_options=DisplayOptions(show={"pre_call_fetch_enabled": [True]}),
),
],
examples=[
NodeExample(
name="warm_greeting",
data={
"name": "Greeting",
"prompt": "Greet warmly and ask the caller's reason for calling.",
"greeting_type": "text",
"greeting": "Hi {{first_name}}, this is Sarah from Acme.",
"allow_interrupt": True,
},
),
],
# `min_outgoing` is intentionally unset: a startCall is allowed to
# sit on the canvas without an outgoing edge (e.g. a workflow with
# just a greeting). Only constraint: nothing flows INTO the start.
graph_constraints=GraphConstraints(
min_incoming=0,
max_incoming=0,
),
)

View file

@ -1,79 +0,0 @@
"""Spec for the API Trigger node — exposes a public webhook URL that
external systems can hit to launch the workflow."""
from api.services.workflow.node_specs._base import (
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertySpec,
PropertyType,
)
SPEC = NodeSpec(
name="trigger",
display_name="API Trigger",
description=("Public HTTP endpoints that launch the workflow."),
llm_hint=(
"Exposes two public HTTP POST endpoints derived from the auto-generated "
"`trigger_path`:\n"
" • Production: `<backend>/api/v1/public/agent/<trigger_path>` — runs "
"the published agent. Use this from production systems.\n"
" • Test: `<backend>/api/v1/public/agent/test/<trigger_path>` — runs "
"the latest draft, useful for verifying changes before publishing. "
"Falls back to the published agent when no draft exists.\n"
"Both require an API key in the `X-API-Key` header.\n"
"Request body fields:\n"
" • `phone_number` (string, required) — destination to dial.\n"
" • `initial_context` (object, optional) — merged into the run's "
"initial context.\n"
" • `telephony_configuration_id` (int, optional) — pick a specific "
"telephony configuration for the call. Must belong to the same "
"organization as the trigger. When omitted, the org's default "
"outbound configuration is used."
),
category=NodeCategory.trigger,
icon="Webhook",
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Name",
description="Short identifier shown in the canvas. No runtime effect.",
required=True,
min_length=1,
default="API Trigger",
),
PropertySpec(
name="enabled",
type=PropertyType.boolean,
display_name="Enabled",
description="When false, the trigger URL returns 404.",
default=True,
),
PropertySpec(
name="trigger_path",
type=PropertyType.string,
display_name="Trigger Path",
description=(
"Auto-generated UUID-style path segment that uniquely "
"identifies this trigger. Used in both URLs:\n"
" • Production: `/api/v1/public/agent/<trigger_path>` — "
"executes the published agent.\n"
" • Test: `/api/v1/public/agent/test/<trigger_path>` — "
"executes the latest draft.\n"
"Do not edit manually."
),
),
],
examples=[
NodeExample(
name="default",
data={"name": "Inbound Trigger", "enabled": True},
),
],
graph_constraints=GraphConstraints(
min_incoming=0,
max_incoming=0,
),
)

View file

@ -1,133 +0,0 @@
"""Spec for the Webhook node — sends an HTTP request to an external system
after the workflow completes."""
from api.services.workflow.node_specs._base import (
GraphConstraints,
NodeCategory,
NodeExample,
NodeSpec,
PropertyOption,
PropertySpec,
PropertyType,
)
SPEC = NodeSpec(
name="webhook",
display_name="Webhook",
description="Send HTTP request after the workflow completes.",
llm_hint=(
"Sends an HTTP request to an external system after the workflow "
"completes. The payload is a Jinja-templated JSON body with access "
"to `workflow_run_id`, `initial_context`, `gathered_context`, "
"`annotations`, and call metadata."
),
category=NodeCategory.integration,
icon="Link2",
properties=[
PropertySpec(
name="name",
type=PropertyType.string,
display_name="Name",
description="Short identifier shown in the canvas and run logs.",
required=True,
min_length=1,
default="Webhook",
),
PropertySpec(
name="enabled",
type=PropertyType.boolean,
display_name="Enabled",
description="When false, the webhook is skipped at run time.",
default=True,
),
PropertySpec(
name="http_method",
type=PropertyType.options,
display_name="HTTP Method",
description="HTTP verb used for the outbound request.",
default="POST",
options=[
PropertyOption(value="GET", label="GET"),
PropertyOption(value="POST", label="POST"),
PropertyOption(value="PUT", label="PUT"),
PropertyOption(value="PATCH", label="PATCH"),
PropertyOption(value="DELETE", label="DELETE"),
],
),
PropertySpec(
name="endpoint_url",
type=PropertyType.url,
display_name="Endpoint URL",
description="URL the request is sent to.",
placeholder="https://api.example.com/webhook",
),
PropertySpec(
name="credential_uuid",
type=PropertyType.credential_ref,
display_name="Authentication",
description="Optional credential applied as the Authorization header.",
llm_hint="Credential UUID from `list_credentials`.",
),
PropertySpec(
name="custom_headers",
type=PropertyType.fixed_collection,
display_name="Custom Headers",
description="Additional HTTP headers to include with the request.",
properties=[
PropertySpec(
name="key",
type=PropertyType.string,
display_name="Header Name",
description="HTTP header name (e.g., 'X-Source').",
required=True,
),
PropertySpec(
name="value",
type=PropertyType.string,
display_name="Header Value",
description="Header value (supports {{template_variables}}).",
required=True,
),
],
),
PropertySpec(
name="payload_template",
type=PropertyType.json,
display_name="Payload Template",
description=(
"JSON body of the request. Values are Jinja-rendered against "
"the run context — `{{workflow_run_id}}`, "
"`{{gathered_context.foo}}`, `{{annotations.qa_xxx}}`, etc."
),
default={
"call_id": "{{workflow_run_id}}",
"first_name": "{{initial_context.first_name}}",
"rsvp": "{{gathered_context.rsvp}}",
"duration": "{{cost_info.call_duration_seconds}}",
"recording_url": "{{recording_url}}",
"transcript_url": "{{transcript_url}}",
},
),
],
examples=[
NodeExample(
name="post_to_crm",
data={
"name": "Notify CRM",
"enabled": True,
"http_method": "POST",
"endpoint_url": "https://crm.example.com/calls",
"payload_template": {
"run_id": "{{workflow_run_id}}",
"outcome": "{{gathered_context.call_disposition}}",
},
},
),
],
# Webhooks fire post-call (run_integrations scans nodes by type),
# never as a graph step. Reject any edge into or out of a webhook so
# the editor can't wire one into the conversation flow.
graph_constraints=GraphConstraints(
min_incoming=0, max_incoming=0, min_outgoing=0, max_outgoing=0
),
)

View file

@ -1,4 +1,5 @@
from typing import TYPE_CHECKING, Awaitable, Callable, Literal, Optional, Union
from typing import TYPE_CHECKING, Awaitable, Callable, Dict, Optional, Union
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import (
@ -17,6 +18,7 @@ from pipecat.services.settings import LLMSettings
from pipecat.utils.enums import EndTaskReason
from api.db import db_client
from api.enums import ToolCategory
from api.services.pipecat.audio_playback import play_audio
from api.services.workflow.disposition_mapper import apply_disposition_mapping
from api.services.workflow.workflow_graph import Node, WorkflowGraph
@ -35,6 +37,7 @@ import asyncio
from loguru import logger
from api.services.workflow import pipecat_engine_callbacks as engine_callbacks
from api.services.workflow.mcp_tool_session import McpToolSession
from api.services.workflow.pipecat_engine_context_composer import (
compose_functions_for_node,
compose_system_prompt_for_node,
@ -117,6 +120,9 @@ class PipecatEngine:
# Cached organization ID (resolved lazily from workflow run)
self._organization_id: Optional[int] = None
# Open MCP tool sessions for this call, keyed by tool_uuid
self._mcp_sessions: Dict[str, McpToolSession] = {}
# Embeddings configuration (passed from run_pipeline.py)
self._embeddings_api_key: Optional[str] = embeddings_api_key
self._embeddings_model: Optional[str] = embeddings_model
@ -179,6 +185,9 @@ class PipecatEngine:
# Helper that encapsulates custom tool management
self._custom_tool_manager = CustomToolManager(self)
# Open persistent MCP server sessions for this call (degrades on failure)
await self._open_mcp_sessions()
# Helper that encapsulates context summarization
if self._context_compaction_enabled:
self._context_summarization_manager = ContextSummarizationManager(self)
@ -504,7 +513,10 @@ class PipecatEngine:
# Register custom tool handlers for this node
if node.tool_uuids and self._custom_tool_manager:
await self._custom_tool_manager.register_handlers(node.tool_uuids)
await self._custom_tool_manager.register_handlers(
node.tool_uuids,
mcp_tool_filters=getattr(node, "mcp_tool_filters", None),
)
# Register knowledge base retrieval handler if node has documents
if node.document_uuids:
@ -530,7 +542,7 @@ class PipecatEngine:
node = self.workflow.nodes[node_id]
logger.debug(
f"Executing node: name: {node.name} is_static: {node.is_static} allow_interrupt: {node.allow_interrupt} is_end: {node.is_end}"
f"Executing node: name: {node.name} allow_interrupt: {node.allow_interrupt} is_end: {node.is_end}"
)
# Track previous node for transition event
@ -585,11 +597,8 @@ class PipecatEngine:
)
await asyncio.sleep(delay_duration)
if node.is_static:
raise ValueError("Static nodes are not supported!")
else:
# Setup LLM Context with Prompts and Functions
await self._setup_llm_context(node)
# Setup LLM context with prompts and functions.
await self._setup_llm_context(node)
def get_node_greeting(self, node_id: str) -> Optional[tuple[str, Optional[str]]]:
"""Return the greeting info for a node, or None if not configured.
@ -685,19 +694,13 @@ class PipecatEngine:
async def _handle_end_node(self, node: Node) -> None:
"""Handle end node execution."""
if node.is_static:
raise ValueError("Static nodes are not supported!")
else:
# Setup LLM Context with Prompts and Functions
await self._setup_llm_context(node)
# Setup LLM context with prompts and functions.
await self._setup_llm_context(node)
async def _handle_agent_node(self, node: Node) -> None:
"""Handle agent node execution."""
if node.is_static:
raise ValueError("Static nodes are not supported!")
else:
# Setup LLM Context with Prompts and Functions
await self._setup_llm_context(node)
# Setup LLM context with prompts and functions.
await self._setup_llm_context(node)
async def end_call_with_reason(
self,
@ -884,6 +887,79 @@ class PipecatEngine:
"""Get the gathered context including extracted variables."""
return self._gathered_context.copy()
async def _open_mcp_sessions(self) -> None:
"""Connect every MCP-category tool referenced by any workflow node.
Failures degrade (session marked unavailable); never raises."""
from api.services.workflow.tools.mcp_tool import (
McpDefinitionError,
validate_mcp_definition,
)
try:
tool_uuids: set[str] = set()
for node in self.workflow.nodes.values():
for tu in getattr(node, "tool_uuids", None) or []:
tool_uuids.add(tu)
if not tool_uuids:
return
organization_id = await self._get_organization_id()
if not organization_id:
logger.warning("Cannot open MCP sessions: organization_id missing")
return
tools = await db_client.get_tools_by_uuids(
list(tool_uuids), organization_id
)
for tool in tools:
if tool.category != ToolCategory.MCP.value:
continue
try:
cfg = validate_mcp_definition(tool.definition)
except McpDefinitionError as e:
logger.warning(
f"Skipping MCP tool '{tool.name}' ({tool.tool_uuid}): "
f"invalid definition: {e}"
)
continue
credential = None
if cfg["credential_uuid"]:
try:
credential = await db_client.get_credential_by_uuid(
cfg["credential_uuid"], organization_id
)
except Exception as e:
logger.warning(
f"MCP tool '{tool.name}': credential fetch failed: {e}"
)
continue
session = McpToolSession(
tool_uuid=tool.tool_uuid,
tool_name=tool.name,
url=cfg["url"],
credential=credential,
tools_filter=cfg["tools_filter"],
timeout_secs=cfg["timeout_secs"],
sse_read_timeout_secs=cfg["sse_read_timeout_secs"],
)
await session.start()
self._mcp_sessions[tool.tool_uuid] = session
except Exception as e:
logger.warning(
f"Failed to open MCP sessions; call proceeds without MCP tools: {e}",
exc_info=True,
)
async def _close_mcp_sessions(self) -> None:
for tool_uuid, session in list(self._mcp_sessions.items()):
try:
await session.close()
except Exception as e:
logger.warning(f"Error closing MCP session {tool_uuid}: {e}")
self._mcp_sessions = {}
async def cleanup(self):
"""Clean up engine resources on disconnect."""
# Cancel any pending timeout tasks
@ -893,6 +969,12 @@ class PipecatEngine:
):
self._user_response_timeout_task.cancel()
# Cancel any in-flight background summarization
if self._context_summarization_manager:
await self._context_summarization_manager.cleanup()
# Cancel any in-flight background summarization.
# MCP sessions are closed in a finally block so they are guaranteed to
# run even if the summarization cleanup raises an exception.
try:
if self._context_summarization_manager:
await self._context_summarization_manager.cleanup()
finally:
# Close any open MCP tool sessions
await self._close_mcp_sessions()

View file

@ -117,7 +117,8 @@ async def compose_functions_for_node(
# Custom tools
if node.tool_uuids and custom_tool_manager:
custom_tool_schemas = await custom_tool_manager.get_tool_schemas(
node.tool_uuids
node.tool_uuids,
mcp_tool_filters=getattr(node, "mcp_tool_filters", None),
)
functions.extend(custom_tool_schemas)

View file

@ -34,6 +34,7 @@ from api.services.workflow.tools.custom_tool import (
)
if TYPE_CHECKING:
from api.services.workflow.mcp_tool_session import McpToolSession
from api.services.workflow.pipecat_engine import PipecatEngine
@ -121,11 +122,18 @@ class CustomToolManager:
"""Get the organization ID from the engine (shared cache)."""
return await self._engine._get_organization_id()
async def get_tool_schemas(self, tool_uuids: list[str]) -> list[FunctionSchema]:
async def get_tool_schemas(
self,
tool_uuids: list[str],
mcp_tool_filters: Optional[dict[str, list[str]]] = None,
) -> list[FunctionSchema]:
"""Fetch custom tools and convert them to function schemas.
Args:
tool_uuids: List of tool UUIDs to fetch
mcp_tool_filters: Optional per-node filter mapping tool_uuid list of
raw MCP tool names to expose. None (default) exposes all tools.
Empty dict or entry with [] suppresses all tools for that uuid.
Returns:
List of FunctionSchema objects for LLM
@ -154,6 +162,22 @@ class CustomToolManager:
)
continue
if tool.category == ToolCategory.MCP.value:
session = self._engine._mcp_sessions.get(tool.tool_uuid)
if session is None or not session.available:
logger.warning(
f"MCP tool '{tool.name}' ({tool.tool_uuid}) "
f"unavailable; skipping"
)
continue
allowed = (
None
if mcp_tool_filters is None
else set(mcp_tool_filters.get(tool.tool_uuid, []))
)
schemas.extend(session.function_schemas(allowed))
continue
raw_schema = tool_to_function_schema(tool)
function_name = raw_schema["function"]["name"]
@ -178,11 +202,18 @@ class CustomToolManager:
logger.error(f"Failed to fetch custom tools: {e}")
return []
async def register_handlers(self, tool_uuids: list[str]) -> None:
async def register_handlers(
self,
tool_uuids: list[str],
mcp_tool_filters: Optional[dict[str, list[str]]] = None,
) -> None:
"""Register custom tool execution handlers with the LLM.
Args:
tool_uuids: List of tool UUIDs to register handlers for
mcp_tool_filters: Optional per-node filter mapping tool_uuid list of
raw MCP tool names to expose. None (default) exposes all tools.
Empty dict or entry with [] suppresses all tools for that uuid.
"""
organization_id = await self.get_organization_id()
if not organization_id:
@ -203,6 +234,32 @@ class CustomToolManager:
)
continue
if tool.category == ToolCategory.MCP.value:
session = self._engine._mcp_sessions.get(tool.tool_uuid)
if session is None or not session.available:
logger.warning(
f"MCP tool '{tool.name}' ({tool.tool_uuid}) "
f"unavailable; skipping handler registration"
)
continue
allowed = (
None
if mcp_tool_filters is None
else set(mcp_tool_filters.get(tool.tool_uuid, []))
)
mcp_schemas = session.function_schemas(allowed)
for fs in mcp_schemas:
self._engine.llm.register_function(
fs.name,
self._create_mcp_handler(session, fs.name),
timeout_secs=session.call_timeout_secs,
)
logger.debug(
f"Registered {len(mcp_schemas)} MCP "
f"handlers for tool '{tool.name}' ({tool.tool_uuid})"
)
continue
schema = tool_to_function_schema(tool)
function_name = schema["function"]["name"]
@ -335,6 +392,29 @@ class CustomToolManager:
return http_tool_handler
def _create_mcp_handler(self, session: "McpToolSession", function_name: str):
"""Create a handler that proxies an LLM function call to a live MCP
session. Errors are returned to the LLM as structured text so the
agent can recover verbally; the call is never crashed."""
async def mcp_tool_handler(
function_call_params: FunctionCallParams,
) -> None:
logger.info(f"MCP Tool EXECUTED: {function_name}")
logger.info(f"Arguments: {function_call_params.arguments}")
try:
result = await session.call(
function_name, function_call_params.arguments or {}
)
await function_call_params.result_callback(result)
except Exception as e:
logger.error(f"MCP tool '{function_name}' failed: {e}")
await function_call_params.result_callback(
{"status": "error", "error": str(e)}
)
return mcp_tool_handler
def _create_end_call_handler(self, tool: Any, function_name: str):
"""Create a handler function for an end call tool.

View file

@ -0,0 +1,116 @@
"""Pure helpers for MCP-category tools: definition validation and
LLM-function-name namespacing. No I/O, no MCP protocol here."""
from __future__ import annotations
import re
from typing import Any, Dict, Literal, Optional
from pydantic import BaseModel, Field, ValidationError, field_validator
DEFAULT_TIMEOUT_SECS = 30
DEFAULT_SSE_READ_TIMEOUT_SECS = 300
class McpDefinitionError(ValueError):
"""Raised when an MCP tool definition is structurally invalid."""
class McpToolConfig(BaseModel):
"""Configuration for an MCP tool definition."""
transport: Literal["streamable_http"] = Field(
default="streamable_http", description="MCP transport protocol"
)
url: str = Field(description="MCP server URL (must be http:// or https://)")
credential_uuid: Optional[str] = Field(
default=None, description="Reference to ExternalCredentialModel for auth"
)
tools_filter: list[str] = Field(
default_factory=list,
description="Allowlist of MCP tool names to expose (empty = all tools)",
)
timeout_secs: int = Field(
default=DEFAULT_TIMEOUT_SECS, description="Connection timeout in seconds"
)
sse_read_timeout_secs: int = Field(
default=DEFAULT_SSE_READ_TIMEOUT_SECS,
description="SSE read timeout in seconds",
)
discovered_tools: list[dict[str, Any]] = Field(
default_factory=list,
description=(
"Server-managed cache of the MCP server's tool catalog "
"[{name, description}]. Populated best-effort by the backend."
),
)
@field_validator("url")
@classmethod
def validate_url(cls, v: str) -> str:
if not isinstance(v, str) or not v.startswith(("http://", "https://")):
raise ValueError("config.url must be an http(s) URL")
return v
@field_validator("tools_filter")
@classmethod
def validate_tools_filter(cls, v: list[str]) -> list[str]:
if not all(isinstance(tool_name, str) for tool_name in v):
raise ValueError("config.tools_filter must be a list of strings")
return v
class McpToolDefinition(BaseModel):
"""Persisted MCP tool definition."""
schema_version: int = Field(default=1, description="Schema version")
type: Literal["mcp"] = Field(description="Tool type")
config: McpToolConfig = Field(description="MCP server configuration")
def _format_validation_error(error: ValidationError) -> str:
parts: list[str] = []
for item in error.errors():
location = ".".join(str(part) for part in item["loc"])
parts.append(f"{location}: {item['msg']}")
return "; ".join(parts)
def validate_mcp_definition(definition: Dict[str, Any]) -> Dict[str, Any]:
"""Validate a ``type: "mcp"`` ToolModel definition and return a
normalized config dict with defaults applied.
Raises:
McpDefinitionError: if the definition is missing required fields
or uses an unsupported transport.
"""
if not isinstance(definition, dict) or definition.get("type") != "mcp":
raise McpDefinitionError("definition.type must be 'mcp'")
config = definition.get("config")
if not isinstance(config, dict):
raise McpDefinitionError("definition.config is required and must be an object")
try:
parsed = McpToolDefinition.model_validate(definition)
except ValidationError as e:
raise McpDefinitionError(_format_validation_error(e)) from e
return parsed.config.model_dump(exclude={"discovered_tools"})
def _slugify(value: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "_", value.strip().lower()).strip("_")
return slug
def namespace_function_name(
tool_name: str, mcp_tool_name: str, *, fallback: str = "server"
) -> str:
"""Build a collision-safe LLM function name: ``mcp__<slug>__<tool>``.
``slug`` is derived from the Dograh ToolModel name; if it slugifies to
empty, ``fallback`` (e.g. first 8 chars of tool_uuid) is used instead.
"""
slug = _slugify(tool_name) or _slugify(fallback) or "server"
return f"mcp__{slug}__{mcp_tool_name}"

View file

@ -4,7 +4,8 @@ from typing import Dict, List, Set
from api.services.workflow.dto import EdgeDataDTO, NodeType, ReactFlowDTO
from api.services.workflow.errors import ItemKind, WorkflowError
from api.services.workflow.node_specs import REGISTRY
from api.services.workflow.node_data import BaseNodeData
from api.services.workflow.node_specs import get_spec
# Regex for matching {{ variable }} template placeholders.
# Captures: group(1) = variable path, group(2) = filter name, group(3) = filter value.
@ -62,7 +63,7 @@ class Edge:
class Node:
def __init__(self, id: str, node_type: NodeType, data):
def __init__(self, id: str, node_type: str, data: BaseNodeData):
self.id, self.node_type, self.data = id, node_type, data
self.out: Dict[str, "Node"] = {} # forward nodes
self.out_edges: List[Edge] = [] # forward edges with properties
@ -75,7 +76,6 @@ class Node:
# Type-specific fields — read with getattr so this works for every
# node variant in the discriminated union.
self.prompt = getattr(data, "prompt", None)
self.is_static = getattr(data, "is_static", False)
self.allow_interrupt = getattr(data, "allow_interrupt", False)
self.extraction_enabled = getattr(data, "extraction_enabled", False)
self.extraction_prompt = getattr(data, "extraction_prompt", None)
@ -84,11 +84,11 @@ class Node:
self.greeting = getattr(data, "greeting", None)
self.greeting_type = getattr(data, "greeting_type", None)
self.greeting_recording_id = getattr(data, "greeting_recording_id", None)
self.detect_voicemail = getattr(data, "detect_voicemail", False)
self.delayed_start = getattr(data, "delayed_start", False)
self.delayed_start_duration = getattr(data, "delayed_start_duration", None)
self.tool_uuids = getattr(data, "tool_uuids", None)
self.document_uuids = getattr(data, "document_uuids", None)
self.mcp_tool_filters = getattr(data, "mcp_tool_filters", None)
self.pre_call_fetch_enabled = getattr(data, "pre_call_fetch_enabled", False)
self.pre_call_fetch_url = getattr(data, "pre_call_fetch_url", None)
self.pre_call_fetch_credential_uuid = getattr(
@ -105,11 +105,11 @@ class WorkflowGraph:
"""
def __init__(self, dto: ReactFlowDTO):
# build adjacency list. n.type comes off the discriminated-union
# variant as a literal string; coerce to NodeType for downstream
# comparisons.
# Build adjacency list from validated DTO nodes. Core node comparisons
# still use NodeType string enums; integration nodes remain plain
# strings and resolve constraints through node specs.
self.nodes: Dict[str, Node] = {
n.id: Node(n.id, NodeType(n.type), n.data) for n in dto.nodes
n.id: Node(n.id, n.type, n.data) for n in dto.nodes
}
# Store all edges
@ -139,7 +139,7 @@ class WorkflowGraph:
# Get a reference to the global node
try:
self.global_node_id = [
n.id for n in dto.nodes if n.type == NodeType.globalNode
n.id for n in dto.nodes if n.type == NodeType.globalNode.value
][0]
except IndexError:
self.global_node_id = None
@ -249,7 +249,7 @@ class WorkflowGraph:
def _assert_global_node(self):
errors: list[WorkflowError] = []
global_node = [
n for n in self.nodes.values() if n.node_type == NodeType.globalNode
n for n in self.nodes.values() if n.node_type == NodeType.globalNode.value
]
if not len(global_node) <= 1:
errors.append(
@ -281,7 +281,7 @@ class WorkflowGraph:
in_deg[m.id] += 1
for n in self.nodes.values():
spec = REGISTRY.get(n.node_type.value)
spec = get_spec(n.node_type)
if spec is None or spec.graph_constraints is None:
continue
gc = spec.graph_constraints