Skip to content

Add a Transcript Source Connector

Before you begin

Roles required: Platform admin, integrator, or developer with access to the Raw to Knowledge codebase.

This topic explains how to add a new pull-based transcript connector to Raw to Knowledge. Use this path when you want Raw to Knowledge to fetch transcripts directly from an external meeting or conversation platform instead of relying on manual POST /v1/artifacts submissions.

This workflow assumes:

  • the source system exposes an API you can call from Raw to Knowledge
  • the source can be transformed into transcript-like segments
  • you want the connector to participate in the standard intake pipeline

Examples of existing connectors in this pattern are Fellow and Gong.

How the connector framework works

Transcript connectors use four layers:

  1. A connector implements the SourceConnector protocol and returns ConnectedTranscript objects.
  2. A connector package exports a provider object from provider.py.
  3. The application discovers providers automatically from raw_to_knowledge.infrastructure.connectors.*.
  4. The ConnectorService ingests fetched transcripts through the normal CII path.

This means a new connector should not require changes to:

  • api/v1/connectors.py
  • route registration
  • the intake pipeline
  • the connector registry builder

If your new connector requires changes in those places, it is not following the intended pattern.

See also: Source Connectors

Steps

1. Create the connector package

Create a new package under:

src/raw_to_knowledge/infrastructure/connectors/{system}/

At minimum, add:

  • __init__.py
  • connector.py
  • provider.py

If the source requires HTTP calls, also add:

  • client.py

Use a short lowercase source_system label such as gong, fellow, or zoom.

2. Implement the connector

In connector.py, implement the SourceConnector protocol.

Your connector must provide:

  • source_system
  • fetch_since()
  • fetch_by_id()
  • close()

The connector returns ConnectedTranscript objects, not SourceArtifact objects. The application layer handles artifact creation and ingestion.

Your connector should transform the external payload into:

{
  "segments": [
    {
      "speaker": "Alice Smith",
      "text": "We should review the rollout plan.",
      "start": "12.0",
      "end": "18.5"
    }
  ]
}

That JSON is the format the existing JSON transcript ingestion path already understands.

3. Export a provider

In provider.py, export a ConnectorProvider object.

This is the discovery hook that allows the connector to be loaded without modifying route or registry code.

Example:

from raw_to_knowledge.infrastructure.connectors.provider import ConnectorProvider
from raw_to_knowledge.infrastructure.connectors.example.connector import ExampleConnector


def _build_example_connector(config: dict[str, str]) -> ExampleConnector:
    return ExampleConnector(
        api_token=config["api_token"],
        base_url=config["base_url"],
    )


provider = ConnectorProvider(
    source_system="example",
    required_config_keys=("api_token",),
    optional_config_keys=("base_url",),
    default_config={"base_url": "https://api.example.com/v1"},
    build_connector=_build_example_connector,
)

4. Configure the connector

New connectors should use CONNECTOR_CONFIGS_JSON.

Example:

{
  "example": {
    "api_token": "secret-token",
    "base_url": "https://api.example.com/v1"
  }
}

Set that JSON in the environment:

export CONNECTOR_CONFIGS_JSON='{"example":{"api_token":"secret-token","base_url":"https://api.example.com/v1"}}'

When the required config keys are present, the connector is registered automatically at runtime.

5. Add tests

Add unit tests under:

tests/unit/connectors/test_{system}_connector.py

Mock the HTTP client at the method level. Do not let tests call the real external API.

You should test:

  • successful fetch_since()
  • bounded fetch_since(since, until=...) behavior if the upstream API supports it
  • successful fetch_by_id()
  • empty-content behavior
  • external API failure behavior
  • transcript transformation
  • source_ref, source_system, and source_type correctness

If the connector introduces new provider behavior, add or extend tests around provider discovery and registry building.

6. Verify discovery

Run the connector test suite:

pytest --no-cov tests/unit/connectors -q

Then start the service with connector config present and verify:

curl http://localhost:8000/v1/connectors

Your new source_system should appear in the response.

7. Configure sync behavior

If the connector should participate in automatic or default incremental pulls, configure sync state:

curl -X PUT http://localhost:8000/v1/connectors/{system}/sync \
  -H "Content-Type: application/json" \
  -d '{
    "enabled": true,
    "cadence": "daily",
    "legal_basis": "legitimate_interest",
    "initial_since": "2026-06-01T00:00:00Z"
  }'

8. Verify pull behavior

Test both connector endpoints:

curl -X POST http://localhost:8000/v1/connectors/{system}/pull \
  -H "Content-Type: application/json" \
  -d '{
    "since": "2026-06-01T00:00:00Z",
    "legal_basis": "legitimate_interest"
  }'

curl -X POST http://localhost:8000/v1/connectors/{system}/fetch/{external_id} \
  -H "Content-Type: application/json" \
  -d '{
    "legal_basis": "legitimate_interest"
  }'

Expected outcomes:

  • GET /v1/connectors lists the connector
  • pull ingests newly fetched transcripts
  • pull without an explicit since uses the configured sync watermark window
  • duplicate pulls are skipped by source_ref idempotency
  • upstream API errors return 502
  • missing connector items return 404

Design rules

Follow these constraints when adding a connector:

  • Keep source-specific API logic in client.py or connector.py, not in API routes.
  • Return ConnectedTranscript, not ORM models or API response dicts.
  • Do not add connector-specific if source_system == ... branches in route code.
  • Prefer CONNECTOR_CONFIGS_JSON over adding new strongly typed settings fields.
  • Preserve provenance: source_ref must be stable and unique for idempotency.
  • If speaker information is available, keep it.
  • If the source has no usable transcript text, skip or reject it explicitly.

Result

The new transcript source is discoverable by convention, configurable without route changes, and ingests through the same governed intake pipeline as every other source.

See also: