chore: clean up dead code

api/services/campaign/campaign_call_dispatcher.py

@@ -20,8 +20,8 @@ from api.utils.common import get_backend_endpoints
 if TYPE_CHECKING:
     # Type-only — importing api.services.telephony eagerly triggers the
     # provider package init, which can pull in this module via the routes
-    # chain and create a circular import. Runtime calls below go through
-    # ``factory.get_telephony_provider`` (lazy import inside the method).
+    # chain and create a circular import. Runtime calls below lazy-import the
+    # factory helpers inside methods instead.
     from api.services.telephony.base import TelephonyProvider
 
 
@@ -31,12 +31,6 @@ class CampaignCallDispatcher:
     def __init__(self):
         self.default_concurrent_limit = int(DEFAULT_ORG_CONCURRENCY_LIMIT)
 
-    async def get_telephony_provider(self, organization_id: int) -> "TelephonyProvider":
-        """Get telephony provider instance for specific organization (default config)."""
-        from api.services.telephony.factory import get_default_telephony_provider
-
-        return await get_default_telephony_provider(organization_id)
-
     async def get_provider_for_campaign(self, campaign) -> "TelephonyProvider":
         """Get the telephony provider pinned to this campaign's config. Falls back
         to the org's default config for legacy campaigns whose

api/services/pricing/workflow_run_cost.py

@@ -27,7 +27,9 @@ async def _fetch_telephony_cost(workflow_run) -> dict | None:
         logger.warning("Workflow not found for workflow run")
         raise Exception("Workflow not found")
 
-    provider = await get_telephony_provider_for_run(workflow_run, workflow.organization_id)
+    provider = await get_telephony_provider_for_run(
+        workflow_run, workflow.organization_id
+    )
     call_cost_info = await provider.get_call_cost(call_id)
 
     if call_cost_info.get("status") == "error":

api/services/telephony/README.md

@@ -13,10 +13,16 @@ Business Logic → TelephonyProvider (Interface) → Concrete Provider (Twilio,
 ### Using the Provider in Code
 
 ```python
-from api.services.telephony.factory import get_telephony_provider
+from api.services.telephony.factory import (
+    get_default_telephony_provider,
+    get_telephony_provider_by_id,
+)
 
-# Get provider based on organization config
-provider = await get_telephony_provider(organization_id)
+# Get the org's default outbound provider
+provider = await get_default_telephony_provider(organization_id)
+
+# Or resolve a specific telephony configuration row
+provider = await get_telephony_provider_by_id(config_id, organization_id)
 
 # Initiate a call
 result = await provider.initiate_call(
@@ -47,7 +53,7 @@ See the [Custom Provider Guide](https://docs.dograh.com/integrations/telephony/c
 
 Quick checklist:
 1. Create `providers/your_provider.py` implementing `TelephonyProvider`
-2. Update `factory.py` to include your provider
+2. Register the package in `providers/__init__.py` and add its schemas to `api/schemas/telephony_config.py`
 3. Write unit tests
 4. Update documentation
 
@@ -107,7 +113,7 @@ class MockProvider(TelephonyProvider):
     # Implement other required methods...
 
 # In tests
-@patch('api.services.telephony.factory.get_telephony_provider')
+@patch('api.services.telephony.factory.get_default_telephony_provider')
 async def test_call_initiation(mock_get_provider):
     mock_get_provider.return_value = MockProvider()
     # Test your business logic
@@ -141,8 +147,8 @@ await service.initiate_call(...)
 
 New code:
 ```python
-from api.services.telephony.factory import get_telephony_provider
-provider = await get_telephony_provider(org_id)
+from api.services.telephony.factory import get_default_telephony_provider
+provider = await get_default_telephony_provider(org_id)
 await provider.initiate_call(...)
 ```
 
@@ -164,4 +170,4 @@ await provider.initiate_call(...)
 - [User Documentation](https://docs.dograh.com/integrations/telephony/overview)
 - [Twilio Integration](https://docs.dograh.com/integrations/telephony/twilio)
 - [Custom Providers](https://docs.dograh.com/integrations/telephony/custom)
-- [Webhooks Guide](https://docs.dograh.com/integrations/telephony/webhooks)
+- [Webhooks Guide](https://docs.dograh.com/integrations/telephony/webhooks)

api/services/telephony/factory.py

@@ -6,9 +6,7 @@ resolution paths exist:
 * by config id — the canonical path used by outbound (test calls, campaigns,
   API triggers) and by the websocket transport once a workflow run has
   ``initial_context.telephony_configuration_id`` stamped on it.
-* by org default — used as a fallback when no specific config is requested
-  (e.g. the legacy ``/telephony-config`` endpoint, the back-compat
-  ``get_telephony_provider(organization_id)`` shim).
+* by org default — used as a fallback when no specific config is requested.
 * for inbound — given a detected provider and an account-id from the webhook,
   iterate the org's configs of that provider and return the one whose stored
   account-id credential matches.
@@ -196,28 +194,6 @@ async def load_credentials_for_transport(
     return config
 
 
-# ---------------------------------------------------------------------------
-# Back-compat shims
-# ---------------------------------------------------------------------------
-
-
-async def load_telephony_config(organization_id: int) -> Dict[str, Any]:
-    """Deprecated: returns the org's default config.
-
-    Existing callers that don't carry a config id continue to work via this
-    shim. New code should pass an explicit telephony_configuration_id."""
-    return await load_default_telephony_config(organization_id)
-
-
-async def get_telephony_provider(organization_id: int) -> TelephonyProvider:
-    """Deprecated: returns a provider for the org's default config.
-
-    See ``load_telephony_config`` above. New code should call
-    ``get_telephony_provider_by_id`` with the resolved config id.
-    """
-    return await get_default_telephony_provider(organization_id)
-
-
 async def get_all_telephony_providers() -> List[Type[TelephonyProvider]]:
     """All registered provider classes — used by inbound webhook detection."""
     return [spec.provider_cls for spec in registry.all_specs()]

api/services/telephony/providers/CLAUDE.md

@@ -28,7 +28,7 @@ If you find yourself editing anything else, re-read the registry plumbing first:
 
 | Want to change... | Source of truth |
 | --- | --- |
-| Outbound provider lookup | `factory.get_telephony_provider*` reads `registry.get(name).provider_cls` |
+| 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` |

api/services/telephony/registry.py

@@ -81,8 +81,8 @@ class ProviderSpec:
             stored config JSON and as the WorkflowRunMode value.
         provider_cls: The TelephonyProvider subclass.
         config_loader: Normalizes raw stored config into the dict shape the
-            provider constructor expects. Replaces the if/elif chain in the
-            old factory.load_telephony_config().
+            provider constructor expects. Replaces the old factory if/elif
+            chain.
         transport_factory: Async callable that creates the pipecat transport
             for an accepted WebSocket. Provider-specific kwargs (stream_sid,
             call_sid, etc.) are forwarded as ``**kwargs``.

docs/integrations/telephony/custom.mdx

@@ -193,7 +193,6 @@ If your provider POSTs webhooks to Dograh (answer URL, status callbacks, hangup
 ```python
 # providers/your_provider/routes.py
 from fastapi import APIRouter, Request
-from api.services.telephony.factory import get_telephony_provider
 from api.services.telephony.status_processor import (
     StatusCallbackRequest,
     _process_status_update,
@@ -286,7 +285,7 @@ register(SPEC)
 | Field | Used by |
 | --- | --- |
 | `name` | Stored as the discriminator on every `TelephonyConfiguration` row and as the `WorkflowRunMode` value |
-| `provider_cls` | `factory.get_telephony_provider*` |
+| `provider_cls` | `factory.get_default_telephony_provider`, `get_telephony_provider_by_id`, `get_telephony_provider_for_run` |
 | `config_loader` | `factory._normalize_with_phone_numbers` (replaces the old if/elif chain) |
 | `transport_factory` | `run_pipeline_telephony` |
 | `audio_config` | `create_audio_config()` and `run_pipeline_telephony` |
@@ -375,7 +374,7 @@ For end-to-end testing, save your provider through the telephony-configurations
 
 ## Best Practices
 
-1. **Trust the registry** — never import another provider's class directly; resolve through `factory.get_telephony_provider*`.
+1. **Trust the registry** — never import another provider's class directly; resolve through the factory helpers (`get_default_telephony_provider`, `get_telephony_provider_by_id`, etc.).
 2. **Sensitive fields** — mark every credential field `sensitive=True` in `ProviderUIMetadata`. The save endpoint masks these on read and preserves the original when the client re-submits a masked value.
 3. **Inbound signature verification** — always validate inbound webhook signatures in `verify_inbound_signature`. Returning `True` when no signature header is present is acceptable; return `False` when a signature *is* present but invalid.
 4. **Transports load credentials lazily** — call `load_credentials_for_transport` with the `telephony_configuration_id` from the workflow run. Don't read the org's default config from `transport.py`.