> ## Documentation Index
> Fetch the complete documentation index at: https://docs.synq.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Building an Integration

> From credentials to a verified graph: the steps a custom integration runs, in order, in Go and Python

A custom integration is a program that reads your tool and writes what it finds to the
Coalesce Quality API. It is **idempotent** — running it twice produces the same estate — so
it is a cron job, not a migration.

Every snippet below is taken from the runnable examples in
[getsynq/api/examples](https://github.com/getsynq/api/tree/main/examples). Pick a language in
any code block and the rest of the page follows.

<Tip>
  The complete program these come from is
  [`bi_tool_lineage`](https://github.com/getsynq/api/tree/main/examples/golang/bi_tool_lineage)
  ([Python](https://github.com/getsynq/api/tree/main/examples/python/bi_tool_lineage)) — a
  whole BI tool modelled end to end, run against a live workspace.
</Tip>

## Credentials

Create client credentials in API settings — `/settings/api` in your own app, which is
`app.synq.io` for EU, `app.us.synq.io` for US and `app.au.synq.io` for AU. Grant these scopes:

| Scope                                                      | Needed for          |
| ---------------------------------------------------------- | ------------------- |
| **Edit Entity Types**                                      | Step 1              |
| **Edit Entities**                                          | Steps 2, 3, 4, 6, 7 |
| **Edit Lineage**                                           | Step 5              |
| **Edit Executions**                                        | Step 8              |
| **Read Entities**, **Read Entity Types**, **Read Lineage** | Steps 1 and 9       |

[Getting Started](/api-reference/getting-started) covers exchanging them for an access token
and generating or installing a client — follow it rather than the short version, it is
generated from the API itself and stays current. Your region's API endpoint is in the
[region table](/security/ip).

What is specific to this guide is the environment it expects and the gRPC channel:

<CodeGroup>
  ```bash Go theme={null}
  export QUALITY_CLIENT_ID=...
  export QUALITY_CLIENT_SECRET=...
  export QUALITY_API_ENDPOINT=...   # your region's API endpoint

  go get buf.build/gen/go/getsynq/api/grpc/go
  go get buf.build/gen/go/getsynq/api/protocolbuffers/go
  ```

  ```bash Python theme={null}
  export QUALITY_CLIENT_ID=...
  export QUALITY_CLIENT_SECRET=...
  export QUALITY_API_ENDPOINT=...   # your region's API endpoint

  pip install grpcio protobuf requests
  pip install getsynq-api-protocolbuffers-python getsynq-api-grpc-python \
    --extra-index-url https://buf.build/gen/python
  ```
</CodeGroup>

Send the token as `authorization: Bearer <token>` on every call and refresh it when
`expires_in` elapses. **Set the gRPC authority explicitly** — without it the port travels in
the `Host` header and every RPC returns `UNIMPLEMENTED` with no message, which reads as a
missing service rather than a routing miss:

<CodeGroup>
  ```go Go theme={null}
  import (
      "golang.org/x/oauth2"
      "golang.org/x/oauth2/clientcredentials"
      "google.golang.org/grpc"
      "google.golang.org/grpc/credentials"
      "google.golang.org/grpc/credentials/oauth"
  )

  cfg := clientcredentials.Config{
      ClientID:     os.Getenv("QUALITY_CLIENT_ID"),
      ClientSecret: os.Getenv("QUALITY_CLIENT_SECRET"),
      TokenURL:     "https://" + endpoint + "/oauth2/token",
  }
  conn, err := grpc.NewClient(endpoint+":443",
      grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{})),
      grpc.WithPerRPCCredentials(oauth.TokenSource{
          TokenSource: cfg.TokenSource(ctx),
      }),
      // Required. Without it the port travels in the Host header and every RPC
      // comes back UNIMPLEMENTED with no message.
      grpc.WithAuthority(endpoint),
  )
  ```

  ```python Python theme={null}
  import grpc, requests, time

  class TokenSource:
      def __init__(self, client_id, client_secret, api_endpoint):
          self.token_url = f"https://{api_endpoint}/oauth2/token"
          self.client_id, self.client_secret = client_id, client_secret
          self.obtain_token()

      def obtain_token(self):
          resp = requests.post(self.token_url, data={
              "client_id": self.client_id,
              "client_secret": self.client_secret,
              "grant_type": "client_credentials",
          })
          resp.raise_for_status()
          self.token = resp.json()
          self.expires_at = time.time() + self.token["expires_in"]

      def get_token(self) -> str:
          if time.time() > self.expires_at:
              self.obtain_token()
          return self.token["access_token"]

  class TokenAuth(grpc.AuthMetadataPlugin):
      def __init__(self, source): self._source = source
      def __call__(self, context, callback):
          callback([("authorization", f"Bearer {self._source.get_token()}")], None)

  channel = grpc.secure_channel(
      f"{endpoint}:443",
      grpc.composite_channel_credentials(
          grpc.ssl_channel_credentials(),
          grpc.metadata_call_credentials(TokenAuth(TokenSource(cid, secret, endpoint))),
      ),
      # Required. Without an explicit authority the port travels in the Host
      # header and every RPC comes back UNIMPLEMENTED with no message.
      options=(("grpc.default_authority", endpoint),),
  )
  ```
</CodeGroup>

## 1. Types

One `UpsertType` per kind of thing your tool has. Declare
[traits](/custom-integrations/modelling#2-entity-types-and-their-traits) — they decide how
your entities rank, sort and behave platform-wide.

<CodeGroup>
  ```go Go theme={null}
  types := []*entitiesv1.Type{
      {
          TypeId:  31,
          Name:    "Metabase Model",
          SvgIcon: iconLayers,
          Traits:  &entitiesv1.TypeTraits{IsModel: proto.Bool(true)},
      },
      {
          TypeId:  32,
          Name:    "Metabase Question",
          SvgIcon: iconChart,
          // A question transforms data AND is read by a person, so it is both.
          Traits: &entitiesv1.TypeTraits{IsModel: proto.Bool(true), IsBiLike: proto.Bool(true)},
      },
      {
          TypeId:  34,
          Name:    "Metabase Alert",
          SvgIcon: iconWarning,
          // A check, not a stage data flows through.
          Traits: &entitiesv1.TypeTraits{IsTestType: proto.Bool(true)},
      },
  }

  for _, t := range types {
      if _, err := api.types.UpsertType(ctx, &customv1.UpsertTypeRequest{Type: t}); err != nil {
          return err
      }
  }
  ```

  ```python Python theme={null}
  types = [
      type_pb2.Type(
          type_id=31,
          name="Metabase Model",
          svg_icon=ICON_LAYERS,
          traits=type_traits_pb2.TypeTraits(is_model=True),
      ),
      type_pb2.Type(
          type_id=32,
          name="Metabase Question",
          svg_icon=ICON_CHART,
          # A question transforms data AND is read by a person, so it is both.
          traits=type_traits_pb2.TypeTraits(is_model=True, is_bi_like=True),
      ),
      type_pb2.Type(
          type_id=34,
          name="Metabase Alert",
          svg_icon=ICON_WARNING,
          # A check, not a stage data flows through.
          traits=type_traits_pb2.TypeTraits(is_test_type=True),
      ),
  ]

  for t in types:
      api.types.UpsertType(types_service_pb2.UpsertTypeRequest(type=t))
  ```
</CodeGroup>

Type numbers are `1`–`1000` and workspace-wide, with no allocator. Call `ListTypes` first and
refuse to start if one of your numbers is taken under a different name — that is the only
protection against two integrations picking the same band.

## 2. Entities

One `UpsertEntity` per object. **Create every entity before anything points at one**: a SQL
binding to an entity that does not exist is rejected at the write, and an entity group whose
members do not all exist is not a set anyone can reconcile against.

<CodeGroup>
  ```go Go theme={null}
  entity := &entitiesv1.Entity{
      Id:          customID("metabase::question::87"),
      TypeId:      32,
      Name:        "Revenue by product category",
      Description: "Revenue split by category, last 90 days.",
      Annotations: []*entitiesv1.Annotation{
          {Name: "metabase.collection", Values: []string{"Commercial"}},
          {Name: "metabase.kind", Values: []string{"question"}},
      },
  }
  _, err := api.entities.UpsertEntity(ctx, &customv1.UpsertEntityRequest{Entity: entity})
  ```

  ```python Python theme={null}
  entity = entity_pb2.Entity(
      id=custom_id("metabase::question::87"),
      type_id=32,
      name="Revenue by product category",
      description="Revenue split by category, last 90 days.",
      annotations=[
          annotation_pb2.Annotation(name="metabase.collection", values=["Commercial"]),
          annotation_pb2.Annotation(name="metabase.kind", values=["question"]),
      ],
  )
  api.entities.UpsertEntity(entities_service_pb2.UpsertEntityRequest(entity=entity))
  ```
</CodeGroup>

Where `customID` builds the identifier:

<CodeGroup>
  ```go Go theme={null}
  func customID(id string) *entitiesv1.Identifier {
      return &entitiesv1.Identifier{
          Id: &entitiesv1.Identifier_Custom{
              Custom: &entitiesv1.CustomIdentifier{Id: id},
          },
      }
  }
  ```

  ```python Python theme={null}
  def custom_id(id_: str) -> identifier_pb2.Identifier:
      return identifier_pb2.Identifier(custom=identifier_pb2.CustomIdentifier(id=id_))
  ```
</CodeGroup>

<Warning>
  **The id is a slug; the readable text belongs in `name`.** Keep it to a lowercase
  `A-Z a-z 0-9 : _ . -` slug built from the source system's stable id. Anything else is folded
  on write — `metabase/question/87` and `metabase question 87` are **the same entity** — and
  ids are case-sensitive, so a capital makes a different one. `name` takes 500 characters of
  any UTF-8, which is where the title, the accents and the spaces go. See
  [Entity ids](/custom-integrations/modelling#1-entity-ids).
</Warning>

To point at an entity from a connected platform, use that platform's own identifier variant
rather than `custom`:

<CodeGroup>
  ```go Go theme={null}
  &entitiesv1.Identifier{
      Id: &entitiesv1.Identifier_BigqueryTable{
          BigqueryTable: &entitiesv1.BigqueryTableIdentifier{
              Project: cfg.project, Dataset: cfg.dataset, Table: "order_items",
          },
      },
  }
  ```

  ```python Python theme={null}
  identifier_pb2.Identifier(
      bigquery_table=identifier_pb2.BigqueryTableIdentifier(
          project=cfg.project, dataset=cfg.dataset, table="order_items",
      )
  )
  ```
</CodeGroup>

Where the platform has no structured variant, or you already hold the entity's opaque
`entity_id`, use the `synq_path` variant instead — see
[Finding the entity\_id](/custom-integrations/modelling#finding-the-entity-id-of-an-existing-entity).

Annotations are your tool's own filing system carried across, so the catalog can be filtered
the way the tool is browsed. Keep them to labels a person would pick out of a list; prose
belongs in the description, which accepts Markdown.

## 3. Schemas

Declare columns **before** any lineage step. A `SELECT *` expands only over an entity whose
columns are known, and a declared column edge only shows against a column the entity has.
Skip this and the table-level edges still appear, with nothing saying the column ones are
missing.

<CodeGroup>
  ```go Go theme={null}
  schema := &customfeaturesv1.Schema{StateAt: timestamppb.Now()}
  for i, c := range columns {
      schema.Columns = append(schema.Columns, &entitiesv1.SchemaColumn{
          Name:            c.name,
          NativeType:      c.nativeType,
          Description:     c.desc,
          OrdinalPosition: int32(i + 1),
      })
  }

  _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
      Feature: &customv1.Feature{
          EntityId: customID("metabase::question::87"),
          // One schema feature per entity, so the id is a constant. Never generate
          // it — a fresh id on every run creates a new feature each time instead
          // of replacing the previous one.
          FeatureId: "schema",
          Feature:   &customv1.Feature_Schema{Schema: schema},
      },
  })
  ```

  ```python Python theme={null}
  schema = schema_pb2.Schema(
      state_at=now(),
      columns=[
          entities_schema_pb2.SchemaColumn(
              name=c.name,
              native_type=c.native_type,
              description=c.desc,
              ordinal_position=i + 1,
          )
          for i, c in enumerate(columns)
      ],
  )

  api.features.UpsertEntityFeature(
      features_service_pb2.UpsertEntityFeatureRequest(
          feature=features_service_pb2.Feature(
              entity_id=custom_id("metabase::question::87"),
              # One schema feature per entity, so the id is a constant. Never
              # generate it — a fresh id on every run creates a new feature each
              # time instead of replacing the previous one.
              feature_id="schema",
              schema=schema,
          )
      )
  )
  ```
</CodeGroup>

## 4. Lineage features

### SqlDefinition, with bindings

The mechanism to reach for first. Give the SQL as the tool executes it, plus the dialect and
a `database_context` for unqualified names.

A name in the SQL that is **not a warehouse object** — another custom entity, a queue, a
service — resolves to nothing and is silently dropped. `references` binds such a name to the
entity it stands for. Names that *are* warehouse objects need no binding.

<CodeGroup>
  ```go Go theme={null}
  _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
      Feature: &customv1.Feature{
          EntityId:  customID("metabase::question::87"),
          FeatureId: "sql",
          Feature: &customv1.Feature_SqlDefinition{
              SqlDefinition: &customfeaturesv1.SqlDefinition{
                  StateAt:         timestamppb.Now(),
                  Dialect:         entitiesv1.SqlDialect_SQL_DIALECT_BIGQUERY,
                  Sql:             question.sql,
                  DatabaseContext: warehouseContext(cfg),
                  References: []*customfeaturesv1.SqlTableReference{{
                      // The name exactly as the SQL writes it. Matched
                      // case-insensitively; leave database_name and schema_name
                      // empty to default them from database_context.
                      ObjectName: "order_items_enriched",
                      Entity:     customID("metabase::model::31"),
                  }},
              },
          },
      },
  })
  ```

  ```python Python theme={null}
  api.features.UpsertEntityFeature(
      features_service_pb2.UpsertEntityFeatureRequest(
          feature=features_service_pb2.Feature(
              entity_id=custom_id("metabase::question::87"),
              feature_id="sql",
              sql_definition=sql_definition_pb2.SqlDefinition(
                  state_at=now(),
                  dialect=sql_dialect_pb2.SQL_DIALECT_BIGQUERY,
                  sql=question.sql,
                  database_context=warehouse_context(cfg),
                  references=[
                      sql_definition_pb2.SqlTableReference(
                          # The name exactly as the SQL writes it. Matched
                          # case-insensitively; leave database_name and schema_name
                          # empty to default them from database_context.
                          object_name="order_items_enriched",
                          entity=custom_id("metabase::model::31"),
                      )
                  ],
              ),
          )
      )
  )
  ```
</CodeGroup>

The bindings are part of the definition and replace with it: one dropped from a later write
is gone, and the lineage it produced is withdrawn.

### ColumnLineage

For a component with no SQL at all. State each edge outright. The declaration is complete
and replaces the previous one whole, so regenerate it from your tool's metadata every run
rather than diffing.

<CodeGroup>
  ```go Go theme={null}
  var edges []*customfeaturesv1.ColumnEdge
  for _, card := range dashboard.cards {
      edges = append(edges, &customfeaturesv1.ColumnEdge{
          Upstream:       customID(fmt.Sprintf("metabase::question::%d", card.fromQuestion)),
          UpstreamColumn: card.fromColumn,
          // The two sides are named independently, so they need not match.
          Column: card.field,
      })
  }

  _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
      Feature: &customv1.Feature{
          EntityId: customID("metabase::dashboard::9"),
          // Only one column-lineage feature per entity, so a stable id is how it
          // is edited.
          FeatureId: "column-lineage",
          Feature: &customv1.Feature_ColumnLineage{
              ColumnLineage: &customfeaturesv1.ColumnLineage{
                  StateAt: timestamppb.Now(),
                  Edges:   edges,
              },
          },
      },
  })
  ```

  ```python Python theme={null}
  edges = [
      column_lineage_pb2.ColumnEdge(
          upstream=custom_id(f"metabase::question::{c.from_question}"),
          upstream_column=c.from_column,
          # The two sides are named independently, so they need not match.
          column=c.field,
      )
      for c in dashboard.cards
  ]

  api.features.UpsertEntityFeature(
      features_service_pb2.UpsertEntityFeatureRequest(
          feature=features_service_pb2.Feature(
              entity_id=custom_id("metabase::dashboard::9"),
              # Only one column-lineage feature per entity, so a stable id is how
              # it is edited.
              feature_id="column-lineage",
              column_lineage=column_lineage_pb2.ColumnLineage(state_at=now(), edges=edges),
          )
      )
  )
  ```
</CodeGroup>

An upstream on a connected platform works the same way, and does not have to exist yet — the
edge is remembered and appears once the entity is ingested. Declaring a column edge also
declares the table edge it implies.

### Code and Git

`Code` carries the SQL, Python, YAML or JSON behind the entity for display; several per
entity. `GitFileReference` says where that code lives, so a connected
[code integration](/code-integrations/overview) can show recent commits against the entity
when it fails.

## 5. Relationships

For edges with nothing to say at column grain. **Read the response** — an edge is held
between the two entities the endpoints *resolve* to, so two relationships that look different
can be the same edge, and only the last write survives.

<CodeGroup>
  ```go Go theme={null}
  resp, err := api.relationships.UpsertRelationships(ctx, &customv1.UpsertRelationshipsRequest{
      Relationships: []*customv1.Relationship{{
          Upstream:   customID("metabase::dashboard::9"),
          Downstream: customID("metabase::subscription::4"),
      }},
  })
  if err != nil {
      return err
  }
  for _, r := range resp.GetResults() {
      fmt.Printf("%s -> %s  %s\n",
          r.GetRelationship().GetUpstream().GetEntityId(),
          r.GetRelationship().GetDownstream().GetEntityId(),
          r.GetOutcome(),
      )
  }
  ```

  ```python Python theme={null}
  resp = api.relationships.UpsertRelationships(
      relationships_service_pb2.UpsertRelationshipsRequest(
          relationships=[
              relationships_service_pb2.Relationship(
                  upstream=custom_id("metabase::dashboard::9"),
                  downstream=custom_id("metabase::subscription::4"),
              )
          ]
      )
  )
  for r in resp.results:
      outcome = relationships_service_pb2.RelationshipWriteOutcome.Name(r.outcome)
      print(f"{r.relationship.upstream.entity_id} -> "
            f"{r.relationship.downstream.entity_id}  {outcome}")
  ```
</CodeGroup>

## 6. Checks

Something that validates another entity is not a stage data flows through. Attach it with a
check relationship, **then** give it a `CheckCategory` feature — attaching schedules no
re-publish of the derived check record, while writing a feature does, so the reverse order
leaves that record stale until the entity is written again.

<CodeGroup>
  ```go Go theme={null}
  _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
      Feature: &customv1.Feature{
          EntityId:  customID("metabase::alert::12"),
          FeatureId: "check",
          Feature: &customv1.Feature_CheckCategory{
              // category and governance_category are left unset on purpose: a
              // producer value outranks the workspace's categorisation rules.
              CheckCategory: &customfeaturesv1.CheckCategory{
                  Package: "metabase",
                  Kind:    alert.kind,
              },
          },
      },
  })

  _, err = api.checkRelationships.UpsertCheckRelationships(ctx,
      &customv1.UpsertCheckRelationshipsRequest{
          CheckRelationships: []*customv1.CheckRelationship{{
              Check:          customID("metabase::alert::12"),
              Checked:        customID("metabase::question::87"),
              CheckedColumns: []string{"revenue"},
          }},
      })
  ```

  ```python Python theme={null}
  api.features.UpsertEntityFeature(
      features_service_pb2.UpsertEntityFeatureRequest(
          feature=features_service_pb2.Feature(
              entity_id=custom_id("metabase::alert::12"),
              feature_id="check",
              # category and governance_category are left unset on purpose: a
              # producer value outranks the workspace's categorisation rules.
              check_category=checks_pb2.CheckCategory(package="metabase", kind=alert.kind),
          )
      )
  )

  api.check_relationships.UpsertCheckRelationships(
      checks_relationships_service_pb2.UpsertCheckRelationshipsRequest(
          check_relationships=[
              checks_relationships_service_pb2.CheckRelationship(
                  check=custom_id("metabase::alert::12"),
                  checked=custom_id("metabase::question::87"),
                  checked_columns=["revenue"],
              )
          ]
      )
  )
  ```
</CodeGroup>

<Warning>
  **Leave `category` and `governance_category` unset.** They are resolved by your workspace's
  own [categorisation rules](/analytics/check-categories), and a value sent by a producer
  outranks those rules. Deriving one from your tool's alert kind — the obvious thing to do —
  silently switches off every rule the workspace wrote, and nothing in the product explains
  why.
</Warning>

## 7. The entity group

One group naming **every entity your integration owns**, every run. The server diffs it
against the previous set and deletes what is missing, so an object deleted in your tool
disappears from the catalog with no state kept on your side.

<CodeGroup>
  ```go Go theme={null}
  ids := everyEntity() // the FULL set, every run
  resp, err := api.groups.UpsertEntitiesGroup(ctx, &customv1.UpsertEntitiesGroupRequest{
      Group: &customv1.Group{GroupId: "metabase", EntityIds: ids},
  })
  if err != nil {
      return err
  }
  for _, deleted := range resp.GetDeletedIds() {
      fmt.Printf("deleted (gone from the tool): %s\n", deleted.GetCustom().GetId())
  }
  ```

  ```python Python theme={null}
  ids = every_entity()  # the FULL set, every run
  resp = api.groups.UpsertEntitiesGroup(
      groups_service_pb2.UpsertEntitiesGroupRequest(
          group=groups_service_pb2.Group(group_id="metabase", entity_ids=ids)
      )
  )
  for deleted in resp.deleted_ids:
      print(f"deleted (gone from the tool): {deleted.custom.id}")
  ```
</CodeGroup>

<Warning>
  **A partial send is a deletion of everything omitted.** If reading your tool can fail
  halfway, skip the group write for that run and let the next complete one reconcile.
</Warning>

<Warning>
  **A group deletes entities only** — not the features, relationships or check relationships of
  entities that still exist. A repointed dashboard, a removed check or a stale `SqlDefinition`
  survives the group write and stays in the graph.
</Warning>

So each run also reconciles what the last one wrote: read the current state with
`ListEntityFeatures` (leave `entity_id` unset for the whole workspace), `ListRelationships` and
`ListCheckRelationships`; diff it against what you are about to write; then call
`DeleteEntityFeature`, `DeleteRelationships` and `DeleteCheckRelationships` for what is gone —
**before** the group write, so no edge is left pointing at an entity that is about to go.

## 8. Executions

What gives a custom entity a status rather than just a position in a graph. `created_at` is
required and must be in the past, so a run that has not happened yet is simply not reported.

<CodeGroup>
  ```go Go theme={null}
  _, err := api.executions.UpsertExecution(ctx, &customv1.UpsertExecutionRequest{
      Execution: &customv1.Execution{
          Id:         customID("metabase::dashboard::9"),
          Status:     customv1.ExecutionStatus_EXECUTION_STATUS_OK,
          Message:    "All cards refreshed.",
          CreatedAt:  timestamppb.New(finished),
          StartedAt:  timestamppb.New(started),
          FinishedAt: timestamppb.New(finished),
      },
  })
  ```

  ```python Python theme={null}
  api.executions.UpsertExecution(
      entity_executions_service_pb2.UpsertExecutionRequest(
          execution=entity_executions_service_pb2.Execution(
              id=custom_id("metabase::dashboard::9"),
              status=entity_executions_service_pb2.EXECUTION_STATUS_OK,
              message="All cards refreshed.",
              created_at=finished,
              started_at=started,
              finished_at=finished,
          )
      )
  )
  ```
</CodeGroup>

Status is `OK`, `WARN`, `ERROR` or `CRITICAL`. `UpsertLogEntry` attaches log output without
changing status.

## 9. Read it back

Every way a custom integration goes wrong produces a valid-looking write and an empty graph,
and the write side cannot tell the difference. End the program by reading its own lineage
back. Lineage is computed asynchronously, so poll.

<CodeGroup>
  ```go Go theme={null}
  depth := int32(10)
  resp, err := api.lineage.GetLineage(ctx, &lineagev1.GetLineageRequest{
      LineageDirection: lineagev1.LineageDirection_LINEAGE_DIRECTION_UPSTREAM,
      StartPoint: &lineagev1.GetLineageStartPoint{
          From: &lineagev1.GetLineageStartPoint_Entities{
              Entities: &lineagev1.EntitiesStartPoint{
                  Entities: []*entitiesv1.Identifier{customID("metabase::dashboard::9")},
              },
          },
      },
      MaxDepth: &depth,
  })
  if err != nil {
      return err
  }
  lin := resp.GetLineage()
  if len(lin.GetNodeDependencies()) == 0 {
      return fmt.Errorf("no upstream lineage: the writes landed but nothing resolved")
  }
  for _, d := range lin.GetNodeDependencies() {
      fmt.Printf("%s -> %s\n",
          nodeName(lin.GetNodes()[d.GetSourceNodeIdx()]),
          nodeName(lin.GetNodes()[d.GetTargetNodeIdx()]),
      )
  }
  ```

  ```python Python theme={null}
  resp = api.lineage.GetLineage(
      lineage_service_pb2.GetLineageRequest(
          lineage_direction=lineage_direction_pb2.LINEAGE_DIRECTION_UPSTREAM,
          start_point=lineage_service_pb2.GetLineageStartPoint(
              entities=lineage_service_pb2.EntitiesStartPoint(
                  entities=[custom_id("metabase::dashboard::9")]
              )
          ),
          max_depth=10,
      )
  )
  lin = resp.lineage
  if not lin.node_dependencies:
      raise RuntimeError("no upstream lineage: the writes landed but nothing resolved")
  for d in lin.node_dependencies:
      print(f"{node_name(lin.nodes[d.source_node_idx])} -> "
            f"{node_name(lin.nodes[d.target_node_idx])}")
  ```
</CodeGroup>

A node carries every identifier that resolves to it, so `nodeName` picks the shape that says
what the node *is* — prefer the `custom` variant when present.

Check for the edges you expect, not merely that some edge came back. A declared column edge
naming a column the upstream does not have still produces one hop while the rest of the chain
is lost, so a non-empty result can still be a broken graph. Poll with a deadline, and exit
non-zero when it expires rather than logging and returning.

For column-level lineage, start from `EntityColumnsStartPoint` instead, naming the entity and
the columns. The response then populates `column_dependencies`, and each node's
`cll_details.cll_state` says whether the parse succeeded. A state of `RESOLUTION_FAILED` on a
node is usually a name your `SqlDefinition` should have bound and did not.

This is not a first-version-only step. Keep it in the production integration.

## Checklist before you ship

* Ids are lowercase slugs from the source system's stable id, in `A-Z a-z 0-9 : _ . -`
* The title and any other human-readable text is in `name`, never in an id
* Every type declares its traits, and its number was checked with `ListTypes` first
* Every entity with columns declares its schema, before any lineage is written
* Every name in a `SqlDefinition` that is not a warehouse object has a binding
* Every upstream resolves, checked with `BatchResolveIdentifiers` (`check_existence: true`)
* Check entities send `package` and `kind`, and leave both categories unset
* One group holds everything, sent whole on every run, never from a partial read
* `feature_id` values are stable strings, not generated per run
* The program reads its own lineage back and fails loudly when it is missing

## Next

<CardGroup cols={2}>
  <Card title="API surface" icon="list" href="/custom-integrations/api-surface">
    Every method, the scope it needs, and the limits
  </Card>

  <Card title="Agent workflow" icon="robot" href="/custom-integrations/agent-workflow">
    A page to point a coding agent at, so it writes the integration for you
  </Card>
</CardGroup>