feat(dq): add DQ rule lifecycle + job observability MCP tools

README.md

@@ -16,6 +16,13 @@ This Go-based MCP server acts as a bridge between AI applications and Collibra,
 - [`get_asset_details`](pkg/tools/get_asset_details/) - Retrieve detailed information about specific assets by UUID, including the asset's assignable attribute schema (every attribute it can hold, including empty ones)
 - [`get_business_term_data`](pkg/tools/get_business_term_data/) - Trace a business term back to its connected physical data assets
 - [`get_column_semantics`](pkg/tools/get_column_semantics/) - Retrieve data attributes, measures, and business assets connected to a column
+- [`get_data_quality_rule`](pkg/tools/get_dq_rule/) - Read the definition of a single DQ rule (monitor) on a job — its type, SQL, filter, tolerance and active/suppressed state
+- [`get_data_quality_rule_results`](pkg/tools/get_dq_rule_results/) - Read a rule's per-run results after a job run — score, breaking/passing record counts, pass/fail status and any exception. Paginated (`offset`/`limit`), newest first by default
+- [`validate_data_quality_rule`](pkg/tools/validate_dq_rule/) - Validate a rule's SQL/definition against the source before saving or running it, so a malformed rule is caught up front. Returns whether the rule is valid plus the engine's message. Requires `edgeSiteId`/`connectionId`/`schemaName` (from `prepare_create_data_quality_job`)
+- [`list_data_quality_rule_templates`](pkg/tools/list_dq_rule_templates/) - List the DQ rule templates (built-in + custom) available in the connected environment via the public DQ API. Each is a parameterized SQL pattern deployable via `deploy_data_quality_rule_template` (by its `ruleTemplateName`). Filters: `name`, `dimension`, `isSystem` (built-in vs custom); paginated
+- [`get_data_quality_rule_template`](pkg/tools/get_dq_rule_template/) - Read a single DQ rule template by `ruleTemplateName` — its parameterized SQL, dialect, dimensions, default tolerance, built-in (`isSystem`) flag and deployed-rule count
+- [`find_data_quality_rules`](pkg/tools/find_dq_rules/) - Search existing DQ rules (monitors) across jobs. Filter by exact `jobName` and/or `columnName` (combine both to detect rules already on a target column) or a rule-name substring. Returns each rule's job, column, type, status and SQL; paginated
+- [`generate_data_quality_rule_sql`](pkg/tools/generate_dq_rule_sql/) - Turn a plain-language description of a check into rule SQL (Text2SQL / Collibra DQ AI), so a rule can be authored without writing SQL. Returns a single SQL string (validate before use). Requires `edgeSiteId`/`connectionId` (from `prepare_create_data_quality_job`)
 - [`get_lineage_downstream`](pkg/tools/get_lineage_downstream/) - Get downstream technical lineage (consumers) for a data entity
 - [`get_lineage_entity`](pkg/tools/get_lineage_entity/) - Get metadata about a specific entity in the technical lineage graph
 - [`get_lineage_transformation`](pkg/tools/get_lineage_transformation/) - Get details and logic of a specific data transformation
@@ -27,6 +34,7 @@ This Go-based MCP server acts as a bridge between AI applications and Collibra,
 - [`prepare_create_asset`](pkg/tools/prepare_create_asset/) - Read-only companion to `create_asset`: enumerate available asset types and domains, resolve a UUID/publicId/displayName for either, and hydrate the scoped attribute and relation schema for a chosen pair
 - [`pull_data_contract_manifest`](pkg/tools/pull_data_contract_manifest/) - Download manifest for a data contract
 - [`search_asset_keyword`](pkg/tools/search_asset_keyword/) - Wildcard keyword search for assets; filters (status, community, domain, domain type, asset type, created-by) accept names or UUIDs
+- [`search_catalog_columns`](pkg/tools/search_catalog_columns/) - Find catalog Column assets by metadata that keyword search can't filter on — Description/Data Type (attribute values), a Data Steward role, or relations to a Business Term/Business Rule/Data Element/Data Attribute (by name); AND-combined. Uses the DGC Knowledge Graph GraphQL API (must be enabled on the instance). Classification-tag filtering is not supported
 - [`search_data_class`](pkg/tools/search_data_classes/) - Search for data classes with filters. **Requires:** `dgc.data-classes-read`
 - [`search_data_classification_match`](pkg/tools/search_data_classification_matches/) - Search for associations between data classes and assets. **Requires:** `dgc.classify`, `dgc.catalog`
 - [`search_lineage_entities`](pkg/tools/search_lineage_entities/) - Search for entities in the technical lineage graph
@@ -37,6 +45,8 @@ This Go-based MCP server acts as a bridge between AI applications and Collibra,
 - [`add_data_classification_match`](pkg/tools/add_data_classification_match/) - Associate a data class with an asset. **Requires:** `dgc.classify`, `dgc.catalog`
 - [`create_assessment`](pkg/tools/create_assessment/) - Conduct a new assessment from a template (given by name or UUID) in the Assessments application. Returns the template's (unanswered) questions to fill in afterward with `edit_assessment` — no separate prepare step needed
 - [`create_asset`](pkg/tools/create_asset/) - Create a new asset of any type. Resolves `assetType` (UUID, publicId, or display name), `domain` (UUID or name), `status` (UUID or name), and attributes (by name or typeId) server-side; converts Markdown to HTML for `RICH_TEXT` attributes; gates on duplicate-name (default `allowDuplicate: false`)
+- [`create_data_quality_rule`](pkg/tools/create_dq_rule/) - Create a data quality rule (monitor) on an existing DQ job. `monitorType` is `FREEFORM_SQL` (full SQL query) or `SIMPLE_SQL` (single-column check); defaults to active and not suppressed. Confirm checkpoint: `confirm=false` (default) returns a preview of the rule + SQL without creating; `confirm=true` creates. Uses the DQ monitoring API and requires permission to create rules on the target job. **Experimental** (`data-quality` feature flag)
+- [`deploy_data_quality_rule_template`](pkg/tools/deploy_dq_rule_template/) - Instantiate a rule template as concrete rules across one or more job/column targets (bulk). The DQ service resolves dialect-specific SQL and names each rule `{templateName}_{columnName}`. Confirm checkpoint: `confirm=false` (default) previews the template + targets without deploying; `confirm=true` deploys. Requires permission to deploy templates and create rules on the target jobs. **Experimental** (`data-quality` feature flag)
 - [`edit_assessment`](pkg/tools/edit_assessment/) - Edit a conducted assessment (identified by name or UUID) via a list of typed operations, applied as a single atomic PATCH (all-or-nothing):
     - `set_answer` - set a question's answer by `questionId`: TEXT/HTML/EXPRESSION/NUMBER/BOOLEAN/DATE via `value`, or ITEMS (choice) via `items`; supply `answerType` for a not-yet-answered question (an already-answered question's type is inferred). ASSETS/USERORGROUPS/ATTACHMENTS answer types are not yet supported
     - `set_status` - move status (`DRAFT`, `SUBMITTED`, `OBSOLETE`)

cmd/chip/experimental.go

@@ -19,7 +19,7 @@ import (
 var knownExperimentalFeatures = map[string]string{
 	skills.FeatureName:                 "Embedded skill catalog served via list_collibra_skills and load_collibra_skill.",
 	tools.ContextSpecificationsFeature: "Context specification tools: list_context_specifications, get_context_specification, and contextSpecificationId parameter on get_asset_details.",
-	tools.DataQualityFeatureName:       "Data Quality job tool (create_data_quality_job: discovery + preview + create in one) that creates and queues DQ jobs.",
+	tools.DataQualityFeatureName:       "Data quality tools: create data quality jobs (create_data_quality_job: discovery + preview + create in one); create, validate, read and search rules; per-run rule results; rule templates (list, read, deploy); plain-language (Text2SQL) rule generation; and catalog column search.",
 }
 
 // validateExperimental warns (without exiting) when the user enabled an

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
-	github.com/rogpeppe/go-internal v1.10.0 // indirect
+	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/sagikazarmark/locafero v0.12.0 // indirect
 	github.com/segmentio/asm v1.1.3 // indirect
 	github.com/segmentio/encoding v0.5.4 // indirect
@@ -26,7 +26,7 @@ require (
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/oauth2 v0.35.0 // indirect
-	golang.org/x/sys v0.41.0 // indirect
-	golang.org/x/text v0.32.0 // indirect
+	golang.org/x/sys v0.45.0 // indirect
+	golang.org/x/text v0.34.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 )

go.sum

@@ -30,8 +30,7 @@ github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0
 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
-github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/sagikazarmark/locafero v0.12.0 h1:/NQhBAkUb4+fH1jivKHWusDYFjMOOKU88eegjfxfHb4=
 github.com/sagikazarmark/locafero v0.12.0/go.mod h1:sZh36u/YSZ918v0Io+U9ogLYQJ9tLLBmM4eneO6WwsI=
 github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc=
@@ -58,10 +57,10 @@ go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ=
 golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA=
-golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
-golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
-golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU=
-golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY=
+golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
+golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=
+golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA=
 golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
 golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

pkg/clients/dgc_kg_client.go

@@ -0,0 +1,211 @@
+package clients
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+)
+
+// This client queries the DGC Knowledge Graph GraphQL API
+// (POST /graphql/knowledgeGraph/v1) to find catalog Column assets by metadata
+// that the public REST search cannot filter on — attribute values, assigned
+// responsibilities (roles), and relations to other assets. It is built against
+// the deployed KG schema (assets(where: AssetFilter)); if a target environment
+// runs a different KG version the filter shapes may differ.
+
+const kgEndpoint = "/graphql/knowledgeGraph/v1"
+
+// Column asset type and the OOTB (is_system) relation-type public ids that link
+// a Column to each catalog asset type, with the direction of the relation
+// relative to the Column and which end holds the other asset.
+const (
+	kgColumnAssetType = "Column"
+
+	// Business Term (Business Asset) --represents--> Column (Data Asset): the
+	// Business Term is the source, so from the Column it is an incoming relation.
+	relBusinessTermPublicID = "BusinessAssetRepresentsDataAsset"
+	// Column (Asset) --governed by--> Business Rule (Governance Asset): Column is
+	// the source, so it is an outgoing relation with the rule as the target.
+	relBusinessRulePublicID = "AssetGovernedByGovernanceAsset"
+	// Data Element --targets--> Data Element (lineage); Column is a Data Element.
+	relDataElementPublicID = "DataElementTargetsDataElement"
+	// Data Attribute --represents--> Column: Data Attribute is the source, so from
+	// the Column it is an incoming relation.
+	relDataAttributePublicID = "DataAttributeRepresentsColumn"
+
+	// Attribute type names used for the value filters.
+	attrDescription = "Description"
+	attrDataType    = "Data Type"
+)
+
+// CatalogColumnSearchParams are the metadata filters, ANDed together. Empty
+// fields are omitted. A specific attribute-type name / relation public-id is
+// applied per field (see constants above).
+type CatalogColumnSearchParams struct {
+	Domain        string
+	Community     string
+	Description   string
+	DataType      string
+	StewardRole   string
+	BusinessTerm  string
+	BusinessRule  string
+	DataElement   string
+	DataAttribute string
+	Limit         int
+	Offset        int
+}
+
+// CatalogColumn is one matching column returned by the search.
+type CatalogColumn struct {
+	ID          string `json:"id"`
+	FullName    string `json:"fullName"`
+	DisplayName string `json:"displayName"`
+	Type        struct {
+		Name string `json:"name"`
+	} `json:"type"`
+	Domain struct {
+		Name string `json:"name"`
+	} `json:"domain"`
+}
+
+type kgSearchResponse struct {
+	Data struct {
+		Assets []CatalogColumn `json:"assets"`
+	} `json:"data"`
+	Errors []struct {
+		Message string `json:"message"`
+	} `json:"errors"`
+}
+
+const kgColumnsQuery = `query Search($where: AssetFilter, $limit: Int, $offset: Int) {
+  assets(where: $where, limit: $limit, offset: $offset) {
+    id
+    fullName
+    displayName
+    type { name }
+    domain { name }
+  }
+}`
+
+// eq builds a StringFilter {eq: v}; contains builds {contains: v}.
+func kgEq(v string) map[string]any       { return map[string]any{"name": map[string]any{"eq": v}} }
+func kgStringEq(v string) map[string]any { return map[string]any{"eq": v} }
+
+// relationFragment builds an incoming/outgoing relation filter fragment matching
+// the other end's displayName. dir is "incomingRelations" or "outgoingRelations";
+// end is "source" or "target" (the end that holds the related asset).
+func relationFragment(dir, publicID, end, value string) map[string]any {
+	return map[string]any{
+		dir: map[string]any{
+			"typePublicId": publicID,
+			"any": map[string]any{
+				end: map[string]any{"displayName": kgStringEq(value)},
+			},
+		},
+	}
+}
+
+// stringAttributeFragment matches a string attribute of the given type name whose
+// value contains v.
+func stringAttributeFragment(attrTypeName, v string) map[string]any {
+	return map[string]any{
+		"stringAttributes": map[string]any{
+			"any": map[string]any{
+				"type":        kgEq(attrTypeName),
+				"stringValue": map[string]any{"contains": v},
+			},
+		},
+	}
+}
+
+// buildWhere assembles the AssetFilter. Fragments that reuse the same top-level
+// key (stringAttributes, incomingRelations, outgoingRelations) cannot coexist in
+// one object, so all fragments are chained through the singular `_and`.
+func buildWhere(p CatalogColumnSearchParams) map[string]any {
+	var frags []map[string]any
+
+	// Always scope to Column.
+	frags = append(frags, map[string]any{"type": kgEq(kgColumnAssetType)})
+
+	if p.Domain != "" || p.Community != "" {
+		domain := map[string]any{}
+		if p.Domain != "" {
+			domain["name"] = kgStringEq(p.Domain)
+		}
+		if p.Community != "" {
+			domain["parent"] = map[string]any{"name": kgStringEq(p.Community)}
+		}
+		frags = append(frags, map[string]any{"domain": domain})
+	}
+	if p.Description != "" {
+		frags = append(frags, stringAttributeFragment(attrDescription, p.Description))
+	}
+	if p.DataType != "" {
+		frags = append(frags, stringAttributeFragment(attrDataType, p.DataType))
+	}
+	if p.StewardRole != "" {
+		frags = append(frags, map[string]any{
+			"responsibilities": map[string]any{
+				"any": map[string]any{"role": kgEq(p.StewardRole)},
+			},
+		})
+	}
+	if p.BusinessTerm != "" {
+		frags = append(frags, relationFragment("incomingRelations", relBusinessTermPublicID, "source", p.BusinessTerm))
+	}
+	if p.BusinessRule != "" {
+		frags = append(frags, relationFragment("outgoingRelations", relBusinessRulePublicID, "target", p.BusinessRule))
+	}
+	if p.DataElement != "" {
+		frags = append(frags, relationFragment("outgoingRelations", relDataElementPublicID, "target", p.DataElement))
+	}
+	if p.DataAttribute != "" {
+		frags = append(frags, relationFragment("incomingRelations", relDataAttributePublicID, "source", p.DataAttribute))
+	}
+
+	// Fold the fragments into a single object chained by `_and`.
+	root := frags[0]
+	cur := root
+	for _, f := range frags[1:] {
+		cur["_and"] = f
+		cur = f
+	}
+	return root
+}
+
+// SearchCatalogColumns finds catalog Column assets matching the given metadata
+// filters via the Knowledge Graph GraphQL API. Requires that KG endpoint to be
+// available on the target instance.
+func SearchCatalogColumns(ctx context.Context, client *http.Client, params CatalogColumnSearchParams) ([]CatalogColumn, error) {
+	limit := params.Limit
+	if limit <= 0 {
+		limit = 25
+	}
+	body := map[string]any{
+		"query": kgColumnsQuery,
+		"variables": map[string]any{
+			"where":  buildWhere(params),
+			"limit":  limit,
+			"offset": params.Offset,
+		},
+	}
+	respBody, status, err := dqDo(ctx, client, http.MethodPost, kgEndpoint, body)
+	if err != nil {
+		return nil, fmt.Errorf("searching catalog columns: %w", err)
+	}
+	if status != http.StatusOK {
+		if status == http.StatusNotFound {
+			return nil, fmt.Errorf("searching catalog columns: knowledge graph endpoint not available on this instance (HTTP 404): %s", string(respBody))
+		}
+		return nil, fmt.Errorf("searching catalog columns: unexpected status %d: %s", status, string(respBody))
+	}
+	var resp kgSearchResponse
+	if err := json.Unmarshal(respBody, &resp); err != nil {
+		return nil, fmt.Errorf("searching catalog columns: decoding response: %w", err)
+	}
+	if len(resp.Errors) > 0 {
+		return nil, fmt.Errorf("searching catalog columns: graphql error: %s", resp.Errors[0].Message)
+	}
+	return resp.Data.Assets, nil
+}

pkg/clients/dq_ai_client.go

@@ -0,0 +1,50 @@
+package clients
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+)
+
+// Text2SQLRequest is the request body for POST /rest/dq/internal/v1/ai/text2sql.
+// It turns a natural-language rule description into rule SQL. The table is
+// identified by jobName; columns give the relevant column context. This endpoint
+// is internal-only (not part of the public DQ API) and is not in the OAS spec.
+type Text2SQLRequest struct {
+	EdgeSiteID   string   `json:"edgeSiteId"`
+	ConnectionID string   `json:"connectionId"`
+	Query        string   `json:"query"`
+	JobName      string   `json:"jobName"`
+	Columns      []string `json:"columns"`
+}
+
+// Text2SQLResponse is the response: a single generated SQL string. There is no
+// separate filter/WHERE-predicate field — the engine returns one query.
+type Text2SQLResponse struct {
+	SQLQuery string `json:"sqlQuery"`
+}
+
+// GenerateDQRuleSQL turns a natural-language description into rule SQL —
+// POST /rest/dq/internal/v1/ai/text2sql.
+func GenerateDQRuleSQL(ctx context.Context, client *http.Client, request Text2SQLRequest) (*Text2SQLResponse, error) {
+	respBody, status, err := dqDo(ctx, client, http.MethodPost, "/rest/dq/internal/v1/ai/text2sql", request)
+	if err != nil {
+		return nil, fmt.Errorf("generating dq rule sql: %w", err)
+	}
+	if status != http.StatusOK {
+		switch status {
+		case http.StatusBadRequest:
+			return nil, fmt.Errorf("generating dq rule sql: bad request (the description could not be turned into valid SQL): %s", string(respBody))
+		case http.StatusForbidden:
+			return nil, fmt.Errorf("generating dq rule sql: missing permission to use DQ AI: %s", string(respBody))
+		default:
+			return nil, fmt.Errorf("generating dq rule sql: unexpected status %d: %s", status, string(respBody))
+		}
+	}
+	var result Text2SQLResponse
+	if err := json.Unmarshal(respBody, &result); err != nil {
+		return nil, fmt.Errorf("generating dq rule sql: decoding response: %w", err)
+	}
+	return &result, nil
+}