How the Disposition Mapper Applies Call Outcome Codes in Dograh

The disposition mapper translates raw call outcome identifiers into organization-specific codes by looking up the DISPOSITION_CODE_MAPPING configuration and returning the mapped value or the original string if no mapping exists.

Dograh's workflow engine normalizes call termination signals—such as "user_idle_max_duration_exceeded" or "voicemail_detected"—into standardized disposition codes like "DAIR" or "VMD" that downstream analytics and CRM systems expect. This translation happens through an async mapping service that consults organization-specific configuration stored in the database. Understanding how the disposition mapper applies these call outcome codes ensures your reporting pipelines receive consistent, normalized data regardless of how individual telephony providers label hang-up reasons.

Core Translation Logic in disposition_mapper.py

The central translation logic resides in api/services/workflow/disposition_mapper.py. The async function apply_disposition_mapping accepts a raw disposition string and an organization ID, then determines the correct output code.

First, it retrieves the optional mapping dictionary by calling db_client.get_configuration_value with the key OrganizationConfigurationKey.DISPOSITION_CODE_MAPPING (lines 22‑27):

# excerpt from api/services/workflow/disposition_mapper.py

mapping = await db_client.get_configuration_value(
    organization_id=organization_id,
    key=OrganizationConfigurationKey.DISPOSITION_CODE_MAPPING
)

If a mapping exists, the function performs a dictionary lookup using mapping.get(value, value), which returns the mapped code when present or falls back to the original value (lines 30‑35). It logs the transformation for audit purposes. Should any exception occur during database access or lookup, the error is caught and logged, and the original raw value is returned to prevent call processing failures (lines 44‑46).

Integration with the Pipecat Engine

The Pipecat engine invokes the mapper when a conversation ends. In api/services/workflow/pipecat_engine.py, the engine first attempts to extract any call_disposition collected during the interaction (lines 66‑68). It then passes this value to apply_disposition_mapping along with the current organization ID (lines 71‑74):

# Inside PipecatEngine._end_call(...)

if call_disposition:
    mapped_disposition = await apply_disposition_mapping(call_disposition, org_id)
    self._gathered_context["mapped_call_disposition"] = mapped_disposition

When no explicit disposition was captured, the engine falls back to the termination reason and maps that instead (lines 80‑84). The final result is stored in the runtime context under the key mapped_call_disposition (lines 76‑78 and 85‑86), making it available for post-call webhooks, analytics ingestion, or external routing logic.

Configuring Organization-Wide Mappings

Admins define permissible translations via the configuration key DISPOSITION_CODE_MAPPING, declared in api/enums.py (line 79). The value expects a JSON object where keys are raw outcome strings and values are the desired downstream codes.

Example configuration:

{
  "user_idle_max_duration_exceeded": "DAIR",
  "no_answer": "NOAN",
  "voicemail_detected": "VMD"
}

When the engine processes a call ending with "user_idle_max_duration_exceeded", the mapper replaces it with "DAIR" before persistence or export.

Practical Implementation Examples

Manual invocation (useful for testing or custom scripts):

from api.services.workflow.disposition_mapper import apply_disposition_mapping
import asyncio

async def demo():
    org_id = 42
    raw = "user_idle_max_duration_exceeded"
    mapped = await apply_disposition_mapping(raw, org_id)
    print(f"Mapped disposition: {mapped}")  # Outputs "DAIR" if configured

asyncio.run(demo())

Pipecat engine integration (simplified excerpt):

call_disposition = self._gathered_context.get("call_disposition", "")
org_id = await self._get_organization_id()

if call_disposition:
    mapped = await apply_disposition_mapping(call_disposition, org_id)
else:
    mapped = await apply_disposition_mapping(reason, org_id)

self._gathered_context["mapped_call_disposition"] = mapped

Summary

• The apply_disposition_mapping function in api/services/workflow/disposition_mapper.py serves as the single source of truth for translating raw outcome strings.
• It fetches the DISPOSITION_CODE_MAPPING configuration via db_client.get_configuration_value and performs a safe dictionary lookup.
• The Pipecat engine automatically invokes the mapper at call termination, storing the result in mapped_call_disposition for downstream consumption.
• Errors during mapping never block execution; the system gracefully returns the original value to ensure call completion reliability.

Frequently Asked Questions

What happens if no disposition mapping is configured for an organization?

If the DISPOSITION_CODE_MAPPING key is absent or empty, apply_disposition_mapping returns the original raw disposition string unchanged. The call outcome is still recorded, but it uses the telephony provider's native identifier rather than a normalized code.

Can the disposition mapper handle fallback values when no explicit call disposition exists?

Yes. The Pipecat engine explicitly checks for empty call_disposition values. When none is found, it passes the termination reason (e.g., "user_disconnected") to the mapper instead, ensuring every call receives a mapped outcome code even when the conversation logic did not explicitly set one.

Where is the mapped disposition stored after translation?

The mapped value is written to the runtime context dictionary under the key mapped_call_disposition within the Pipecat engine's _gathered_context. This makes it available to post-call processors, webhook payloads, and analytics exporters immediately after the mapping function completes.

How do I add a new call outcome code mapping?

Define the raw identifier and desired code in the JSON value for OrganizationConfigurationKey.DISPOSITION_CODE_MAPPING via your organization's configuration settings. For example, adding "busy_signal": "BUSY" to the JSON object will cause the mapper to translate raw "busy_signal" outcomes to "BUSY" for all future calls in that organization.

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