First pipeline

This example copies two orders from one PostgreSQL table into another. It gives you a result you can verify without a transform, ontology package or model. It uses the Postgres read and write capabilities and two workspace endpoints in one project.

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.

Use a reachable test PostgreSQL database, its SQL client and a database identity authorized to create/read the source fixture and write the dedicated output. Programmatic creation uses pipelines:write; execution uses pipelines:execute; run inspection uses runs:read. Endpoint creation and plugin installation need their own permissions.

Prepare the source

Run this in a test database. Use a fresh table name if the table already exists, and update the endpoint target accordingly.

CREATE TABLE public.pipeline_demo_orders (
  order_id integer PRIMARY KEY,
  region text NOT NULL,
  total numeric(12,2) NOT NULL
);
INSERT INTO public.pipeline_demo_orders VALUES
  (1, 'North', 120.00),
  (2, 'South', 75.50);

Open workspace Plugins → Explore, find Postgres (@craven/postgres), inspect its installed contract and install the read capability. Fill these connection fields and run its supported connection check:

Open workspace Data Endpoints → New data endpoint. Use name demo_orders, namespace pipelines, direction Ingress, the installed Postgres read capability and Stream (source_stream) as the source contract. In the target fields set Schema to public and Table to pipeline_demo_orders. Save and inspect its schema/sample. demo_orders is a name you choose in Semogram; public.pipeline_demo_orders is the actual database table. Keep its returned endpoint UUID for programmatic examples.

The Postgres installation guide and endpoint guide provide additional detail; the required setup for this fixture is included here.

Prepare the destination

In Plugins → Explore, install Postgres's write capability with the same connection fields above and an identity authorized for the test output. Read and write can use different database users. The writer can perform table/schema setup and enable row-level security; give it the needed setup privileges on this dedicated test target.

Create the output table with your SQL client:

CREATE TABLE public.pipeline_demo_orders_copy (
  order_id integer PRIMARY KEY,
  region text NOT NULL,
  total numeric(12,2) NOT NULL
);

Create a second workspace endpoint: name demo_orders_copy, namespace pipelines, direction Egress, installed Postgres write capability, contract write, target Schema public, Table pipeline_demo_orders_copy, Write mode append. Save it. It is a destination, not another source endpoint. Keep its UUID for programmatic creation.

Create the graph

Create a pipeline in this project named Copy two demo orders. Read the workspace endpoint demo_orders in full mode using parser extraction. Connect that output to an egress node using demo_orders_copy in append mode. Do not add other data sources or a schedule. Show me the two endpoint targets and proposed graph before saving.

curl --request POST "https://platform.semogram.com/api/v1/projects/<PROJECT_ID_UUID>/pipelines" \
  --header "Authorization: Bearer ${SEMOGRAM_API_KEY}" \
  --header "Idempotency-Key: demo-order-copy-create-001" \
  --header "Content-Type: application/json" \
  --data-binary @pipeline-create.json

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "pipeline_create",
    "arguments": {
      "projectId": "22222222-2222-4222-8222-222222222222",
      "document": {
        "version": 1,
        "kind": "studio-plan",
        "metadata": {
          "name": "Copy two demo orders",
          "description": "Copy the two-row test fixture into a dedicated output table",
          "projectId": "22222222-2222-4222-8222-222222222222",
          "workspaceId": "11111111-1111-4111-8111-111111111111"
        },
        "graph": {
          "id": "demo-order-copy",
          "nodes": [
            {
              "id": "read-orders",
              "kind": "ingress",
              "label": "Read demo orders",
              "inputs": [],
              "outputs": [
                {
                  "key": "out",
                  "label": "Orders",
                  "direction": "output",
                  "interface": {
                    "shape": "collection"
                  }
                }
              ],
              "policy": {},
              "ingress": {
                "dataEndpointId": "33333333-3333-4333-8333-333333333333",
                "mode": "full",
                "output": "orders",
                "strategy": "parser"
              }
            },
            {
              "id": "write-orders",
              "kind": "egress",
              "label": "Write demo orders",
              "inputs": [
                {
                  "key": "in",
                  "label": "Orders",
                  "direction": "input",
                  "interface": {
                    "shape": "collection"
                  }
                }
              ],
              "outputs": [],
              "policy": {},
              "egress": {
                "dataEndpointId": "44444444-4444-4444-8444-444444444444",
                "source": "orders",
                "writeMode": "append"
              }
            }
          ],
          "edges": [
            {
              "id": "orders-to-output",
              "fromNodeId": "read-orders",
              "fromOutput": "out",
              "toNodeId": "write-orders",
              "toInput": "in"
            }
          ]
        }
      },
      "commitMessage": "Initial two-row fixture",
      "idempotencyKey": "demo-order-copy-create-001"
    }
  }
}

Validate and save

In Studio, Validate the draft. Resolve errors, inspect warnings and Save version after edits. Confirm the selected version contains these two endpoints and append mode. Preflight does not copy either row or test the external write.

API clients can validate the actor's draft with POST /api/v1/projects/<PROJECT_ID>/pipelines/<PIPELINE_ID>/draft/validate (pipelines:read) and commit the full document with POST .../versions (pipelines:write). MCP uses pipeline_validate with pipelineId and projectId; saving a draft through pipeline_draft_save does not commit a version. The initial version returned by creation can be inspected before launch.

Run once

Validate Copy two demo orders and show the active saved version and both targets. After I approve the write, run that saved pipeline once. Report the run ID and inspect its final step results; do not schedule it.

curl --request POST "https://platform.semogram.com/api/v1/projects/<PROJECT_ID_UUID>/pipelines/<PIPELINE_ID_UUID>/execute" \
  --header "Authorization: Bearer ${SEMOGRAM_API_KEY}" \
  --header "Idempotency-Key: pipeline-demo-operation-001"

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "pipeline_execute",
    "arguments": {
      "pipelineId": "<PIPELINE_ID_UUID>",
      "projectId": "<PROJECT_ID_UUID>",
      "idempotencyKey": "demo-order-copy-run-001"
    }
  }
}

Keep runId from the accepted response. Poll GET /api/v1/projects/<PROJECT_ID>/runs/<RUN_ID> with runs:read, or call MCP job_get with jobId set to that run ID and projectId. Disconnecting your client does not make a queued run disappear.

Check the result

Use your database SQL client:

SELECT order_id, region, total
FROM public.pipeline_demo_orders_copy
ORDER BY order_id;

Inspect the source too: it should still contain the two original rows. Check that the run's ingress and egress steps succeeded and inspect their errors/counts instead of accepting only an assistant summary.

Append does not turn a repeat into an upsert. With these primary keys, rerunning can fail on duplicate keys; a failed write may have committed some rows. Inspect the output before retrying. To repeat this fixture, empty only the dedicated test output using your database tools and run once again. Do not clear a shared business table.

If something fails

Once this fixture works, try Transform records. Do not enable a recurring schedule until the expected behavior and replay policy are established.