Custom plugin pipeline

Run a custom read capability through an endpoint without changing the pipeline model

A pipeline can use a custom plugin through the same installed-capability and endpoint model as an internal connector. This example reads two shipment-event records from a private test service. The custom plugin supplies the implementation; the endpoint selects its target; the pipeline orchestrates the read.

What you need

You need a Semogram account with workspace membership, a project in that workspace and permission to create and run pipelines. Reading or writing also requires access to the selected endpoints and external systems. Creating a pipeline does not grant those permissions.

Your team must have a real packaged read capability implementing source_stream, a reachable test service and its account credentials. The example contract below is custom, not a built-in Semogram source type. A pipeline definition cannot implement an unfinished connector.

Define and test the fixture contract

The test service must return these two records for account demo_account and resource shipment_events:

Expected service fixture

[
  {
    "event_id": "evt-1",
    "shipment_id": "ship-1",
    "status": "departed"
  },
  {
    "event_id": "evt-2",
    "shipment_id": "ship-2",
    "status": "arrived"
  }
]

The custom package should declare protected installation fields baseUrl and token, target fields accountId and resource, and actual bounded reads for that selector. Implement pagination, cursor handling, errors and telemetry according to its contract. Declare schema discovery/incremental support only if implemented. Test the service and packaged connector against the fixture before publishing.

Publish/register the versioned package through your supported custom plugin workflow, then open workspace Plugins → Explore, inspect its published manifest and install its read capability. Fill the actual base URL and protected token field; check the connection. The custom plugin guide has package-authoring detail, but this pipeline scenario assumes that implementation has passed its fixture test.

Create the endpoint

Open workspace Data Endpoints → New data endpoint. Name it demo_shipments, namespace pipelines, direction Ingress, select the installed custom read capability and source_stream contract. Fill target fields Account id demo_account and Resource shipment_events. These labels come from this custom schema. Check the selected account/resource, schema and two returned rows.

Create and execute the pipeline

Assistant prompt

Create a pipeline reading demo_shipments in full mode with parser extraction and output shipments. Use the installed custom read capability through that endpoint. Show its account/resource target and the two expected event fields before saving. Do not assume incremental support or add a write target.

Create a pipeline in the project. Select an Ingress node, choose Data endpoint demo_shipments, Load mode Full load and parser extraction. Inspect its selected capability and target. Use an inspectable intermediate output if needed; no egress is required to establish the read fixture.

ingress section

{
  "dataEndpointId": "<CUSTOM_SOURCE_ENDPOINT_UUID>",
  "mode": "full",
  "output": "shipments",
  "strategy": "parser"
}

Validate and save a version, then run once. Inspect the ingress artifact/output and compare both event IDs, shipment IDs and statuses with the service fixture. Confirm the run identifies the installed package version. Test the connector's error and paging cases before scheduling.

A custom transform would instead bind its installed transform capability directly on a Transform node. A custom writer needs a separate implemented write capability and Egress endpoint. The package’s declared/read-tested contract does not establish either of those behaviors.

What you need

Define and test the fixture contract

Create the endpoint

Create and execute the pipeline

On this page