cribl-stream

name	cribl-stream
description	Build MCP servers, author routes/pipelines/functions, and manage all Cribl Stream artifacts (Stream, Edge, Search, Lake) via REST API and YAML configs. Use when user mentions Cribl, Cribl Stream/Edge/Search/Lake, routes, pipelines, sources, destinations, packs, or observability pipeline management.
metadata	{"author":"kundeng","version":"1.0.0","source":"Cribl Stream Documentation Manual v4.17 (generated 2026-04-22)"}

Cribl Stream Skill

You are an expert in Cribl Stream — the observability pipeline platform for routing, transforming, and delivering machine data. You help users build MCP servers that manage Cribl Stream artifacts and author configurations programmatically.

Online Documentation — Fetch When Needed

This skill captures the v4.17.1 docs offline, but versions, endpoint shapes, and new artifact types evolve. When the local reference here is incomplete, ambiguous, or you suspect it's stale, fetch the canonical online docs.

Authoritative URLs (use WebFetch or the docs MCP)

Resource	URL
Cribl Stream docs (current)	https://docs.cribl.io/stream/
Cribl Stream API Reference (Swagger)	https://docs.cribl.io/api-reference/
Cribl Stream OpenAPI spec	https://docs.cribl.io/stream/api-reference/
Cribl Edge docs	https://docs.cribl.io/edge/
Cribl Search docs	https://docs.cribl.io/search/
Cribl Lake docs	https://docs.cribl.io/lake/
Cribl.Cloud docs	https://docs.cribl.io/cloud/
Routes deep-dive	https://docs.cribl.io/stream/routes/
Pipelines deep-dive	https://docs.cribl.io/stream/pipelines/
Functions reference	https://docs.cribl.io/stream/functions/
Sources reference	https://docs.cribl.io/stream/sources/
Destinations reference	https://docs.cribl.io/stream/destinations/
Packs Dispensary	https://packs.cribl.io/
GitOps & Version Control	https://docs.cribl.io/stream/gitops/
REST/API Endpoint Collector	https://docs.cribl.io/stream/collectors-rest/
CLI reference	https://docs.cribl.io/stream/cli-reference/
Config files reference	https://docs.cribl.io/stream/config-files/
Cribl Expressions (C.*)	https://docs.cribl.io/stream/cribl-expressions/
Knowledge Objects	https://docs.cribl.io/stream/knowledge-objects/
Worker Groups	https://docs.cribl.io/stream/worker-groups/
Notifications	https://docs.cribl.io/stream/notifications/
Community Slack	https://cribl.io/community/
GitHub (Cribl org)	https://github.com/criblio
Cribl Outpost	https://docs.cribl.io/stream/outpost/

When to fetch online

• Building/calling endpoints not listed below → fetch the API Reference for exact request/response shape
• Configuring a Source/Destination type with vendor-specific fields (e.g., a new SaaS connector, OAuth quirks) → fetch the per-destination docs
• User is on a different Cribl version (3.x, newer than 4.17, or Cribl.Cloud) → confirm endpoint paths haven't changed
• A YAML field name doesn't match what you remember → fetch the matching config file reference page
• Building a Pack for the Dispensary → fetch packs.cribl.io for current publication standards

Prefer Context7 MCP (resolve-library-id for cribl) when available, otherwise use WebFetch directly. Don't guess endpoint paths or field names — fetch and confirm.

When to Use This Skill

• User wants to build an MCP server for Cribl Stream management
• User needs to author/modify routes, pipelines, functions, sources, destinations, packs
• User asks about the Cribl Stream REST API
• User needs to automate Cribl Stream configuration via code
• User asks about Cribl config file formats (YAML)

---

Cribl Stream Architecture Overview

Cribl Stream processes machine data in real time. The data flow is:

Sources → [Pre-Processing Pipeline] → Routes → Processing Pipelines → [Post-Processing Pipeline] → Destinations

Core Concepts

• Sources (inputs.yml): Data ingestion points (Syslog, Kafka, S3, Splunk HEC, HTTP, etc.)
• Routes (pipelines/route.yml): Filter expressions that select events and direct them to Pipelines. Evaluated top-down. Each route has a filter (JS expression), a pipeline, a destination, and a Final toggle.
• Pipelines (pipelines//conf.yml): Ordered sequences of Functions that transform events.
• Functions: Individual processing steps within pipelines (Eval, Mask, Drop, Lookup, Parser, Regex Extract, etc.)
• Destinations (outputs.yml): Where processed data is sent (Splunk, S3, Elasticsearch, Kafka, Syslog, etc.)
• Packs: Self-contained bundles of routes, pipelines, sources, destinations, and knowledge objects.
• Knowledge Objects: Reusable config — Variables, Lookups, Event Breakers, Regexes, Parsers, Schemas, Grok Patterns.
• Worker Groups: In distributed deployments, groups of worker nodes with shared configuration.

---

Cribl Stream REST API Reference

Authentication

# Login — returns a Bearer token
POST /api/v1/auth/login
Content-Type: application/json
Body: {"username": "<user>", "password": "<password>"}
# Response: {"token": "..."}

# Use the token in subsequent requests:
Authorization: Bearer <token>

The API server listens on port 9000 by default (configurable in cribl.yml). Token TTL defaults to 3600 seconds (configurable via api.authTokenTTL in cribl.yml).

API URL Patterns

Base URL: http://<host>:9000/api/v1

For distributed deployments (Worker Groups):

• GET /api/v1/m/<groupName>/routes — routes for a specific worker group
• GET /api/v1/m/<groupName>/pipelines — pipelines for a group
• GET /api/v1/m/<groupName>/inputs — sources for a group
• GET /api/v1/m/<groupName>/outputs — destinations for a group

For single-instance deployments, omit the /m/<groupName> segment.

Key API Endpoints

Routes

GET    /api/v1/routes                    — List all routes
GET    /api/v1/routes/<id>               — Get a specific route
POST   /api/v1/routes                    — Create a route
PATCH  /api/v1/routes/<id>               — Update a route
DELETE /api/v1/routes/<id>               — Delete a route

Pipelines

GET    /api/v1/pipelines                 — List all pipelines
GET    /api/v1/pipelines/<id>            — Get a specific pipeline
POST   /api/v1/pipelines                 — Create a pipeline
PATCH  /api/v1/pipelines/<id>            — Update a pipeline
DELETE /api/v1/pipelines/<id>            — Delete a pipeline

Sources (Inputs)

GET    /api/v1/inputs                    — List all sources
GET    /api/v1/inputs/<id>              — Get a specific source
POST   /api/v1/inputs                    — Create a source
PATCH  /api/v1/inputs/<id>              — Update a source
DELETE /api/v1/inputs/<id>              — Delete a source

Destinations (Outputs)

GET    /api/v1/outputs                   — List all destinations
GET    /api/v1/outputs/<id>             — Get a specific destination
POST   /api/v1/outputs                   — Create a destination
PATCH  /api/v1/outputs/<id>             — Update a destination
DELETE /api/v1/outputs/<id>             — Delete a destination

Packs

GET    /api/v1/packs                     — List all packs
POST   /api/v1/packs                     — Install a pack
DELETE /api/v1/packs/<id>               — Remove a pack

Knowledge Objects

GET    /api/v1/lib/vars                  — List global variables
POST   /api/v1/lib/vars                  — Create a variable
PATCH  /api/v1/lib/vars/<id>            — Update a variable
DELETE /api/v1/lib/vars/<id>            — Delete a variable

GET    /api/v1/lib/lookups               — List lookups
GET    /api/v1/lib/regexes               — List regexes
GET    /api/v1/lib/breakers              — List event breakers
GET    /api/v1/lib/parsers               — List parsers
GET    /api/v1/lib/schemas               — List JSON schemas
GET    /api/v1/lib/parquet-schemas       — List Parquet schemas

Version Control & Deployment

POST   /api/v1/version/commit            — Commit changes
POST   /api/v1/version/deploy            — Deploy committed changes
POST   /api/v1/version/sync             — Sync from Git remote (GitOps)
GET    /api/v1/version/info              — Get version info

System

GET    /api/v1/system/info               — System information
GET    /api/v1/system/metrics            — System metrics
GET    /api/v1/system/logs               — System logs
POST   /api/v1/system/restart            — Restart Cribl Stream
POST   /api/v1/system/reload             — Reload configuration
GET    /api/v1/health                    — Health check

Workers (Distributed)

GET    /api/v1/master/groups             — List worker groups
GET    /api/v1/master/workers            — List all workers
GET    /api/v1/master/groups/<id>/workers — Workers in a group

Jobs (Collectors)

GET    /api/v1/jobs                      — List jobs
POST   /api/v1/jobs                      — Create/run a job
GET    /api/v1/jobs/<id>                — Get job status
DELETE /api/v1/jobs/<id>                — Cancel a job

Notifications

GET    /api/v1/notifications             — List notifications
POST   /api/v1/notifications             — Create a notification

Secrets

GET    /api/v1/secrets                   — List secrets
POST   /api/v1/secrets                   — Create a secret

---

Configuration File Formats

All config files are YAML. Located at $CRIBL_HOME/local/cribl/ (single-instance) or $CRIBL_HOME/groups/<groupName>/local/cribl/ (distributed).

Route Configuration (pipelines/route.yml)

id: default
groups: {}
comments: []
routes:
  - id: my_route_name
    name: my_route_name
    filter: "source=='syslog' && severity<=3"
    pipeline: my_pipeline
    output: splunk_hec
    final: true
    disabled: false
    description: "Route critical syslog to Splunk"
  - id: catch_all
    name: catch_all
    filter: "true"
    pipeline: passthru
    output: default
    final: true

Route fields:

• id / name: Unique identifier (case-sensitive)
• filter: JavaScript expression evaluated per event. true matches all events.
• pipeline: Pipeline ID or Pack reference to process matched events
• output: Destination ID to send processed events to
• final: When true (default), matched events stop at this route. When false, events continue to next route (cloning behavior).
• disabled: Toggle route on/off
• description: Optional description
• cloneFields: When final=false, additional fields to add to cloned events

Pipeline Configuration (pipelines//conf.yml)

output: default
groups: {}
asyncFuncTimeout: 1000
functions:
  - id: eval
    filter: "true"
    disabled: false
    conf:
      add:
        - name: datacenter
          value: "'us-east-1'"
      remove:
        - date_*
  - id: mask
    filter: "sourcetype=='access_combined'"
    disabled: false
    conf:
      rules:
        - matchRegex: "/(\\d{3}-\\d{2}-)\\d{4}/g"
          replaceExpr: "`${g1}XXXX`"
      fields:
        - _raw
  - id: drop
    filter: "severity > 6"
    disabled: false
    final: true

Source Configuration (inputs.yml)

inputs:
  syslog_input:
    type: syslog
    host: "0.0.0.0"
    port: 514
    protocol: udp
    sendToRoutes: true
    pipeline: pre_process_syslog
    disabled: false
    metadata:
      - name: source
        value: "'syslog'"
    breakerRulesets:
      - Syslog

  splunk_hec_input:
    type: splunk_hec
    host: "0.0.0.0"
    port: 8088
    token: "my-hec-token"
    sendToRoutes: true
    disabled: false

  http_input:
    type: http
    host: "0.0.0.0"
    port: 10080
    sendToRoutes: true

Destination Configuration (outputs.yml)

outputs:
  splunk_hec:
    type: splunk_hec
    url: "https://splunk.example.com:8088/services/collector/event"
    token: "my-hec-token"
    compress: true
    onBackpressure: block
    pipeline: post_process_splunk

  s3_archive:
    type: s3
    bucket: "my-archive-bucket"
    region: "us-east-1"
    awsAuthenticationMethod: auto
    stagePath: "/tmp/cribl/s3"
    compress: gzip
    partitionExpr: "`${sourcetype}/${_time.strftime('%Y/%m/%d')}`"

  devnull:
    type: devnull

  default:
    type: default
    defaultId: splunk_hec

  webhook_out:
    type: webhook
    url: "https://webhook.example.com/events"
    method: POST
    format: json
    authType: token
    token: "bearer-token-here"

  syslog_out:
    type: syslog
    host: "syslog.example.com"
    port: 514
    protocol: tcp
    facility: 1
    severity: 6
    messageFormat: rfc5424

  elasticsearch_out:
    type: elastic
    url: "https://es.example.com:9200"
    index: "logs-cribl"
    docTypeExpr: "'_doc'"

Global Variables (vars.yml)

variables:
  my_threshold:
    type: number
    value: 42
    description: "Event threshold for alerting"
  api_key:
    type: encrypted_string
    value: "encrypted_value_here"
    description: "External API key"
  blocked_hosts:
    type: array
    value: '["bad-host-1", "test-server-99"]'

Reference in expressions: C.vars.my_threshold, C.vars.blocked_hosts.includes(host)

Event Breakers (breakers.yml)

id: my_breaker_ruleset
lib: custom
description: "Custom event breakers"
rules:
  - name: json_events
    condition: "/^\\{/"
    type: json
    timestampAnchorRegex: "/\"timestamp\"\\s*:\\s*/"
    timestamp:
      type: auto
  - name: syslog_events
    condition: "/^<\\d+>/"
    type: regex
    timestampAnchorRegex: "/^/"
    eventBreakerRegex: "/[\\r\\n]+/"

---

Available Functions (Pipeline Processing Steps)

Function	Purpose
Eval	Add, modify, or remove event fields
Mask	Regex-based pattern masking/redaction (supports C.Mask.random, C.Mask.md5, C.Mask.sha256, C.Mask.REDACTED, etc.)
Drop	Drop events matching a filter
Sampling	Sample events at a configurable rate
Dynamic Sampling	Sample based on field cardinality
Regex Extract	Extract fields using regex capture groups
Regex Filter	Filter events based on regex patterns
Parser	Parse structured data (JSON, KV, CSV, etc.)
Serialize	Serialize events to a specific format
Lookup	Enrich events from lookup tables
GeoIP	Add geographic data from IP addresses
DNS Lookup	Resolve hostnames/IPs
Clone	Duplicate events with modifications
Drop Dimensions	Remove metric dimensions to reduce cardinality
Flatten	Flatten nested objects
Fold Keys	Fold object keys matching patterns
Rename	Rename fields
Unroll	Expand array fields into separate events
JSON Unroll	Unroll JSON arrays
XML Unroll	Unroll XML elements
Aggregations	Aggregate events over time windows
Aggregate Metrics	Aggregate metric events
Publish Metrics	Convert events to metrics format
Rollup Metrics	Roll up metric granularity
Suppress	Suppress duplicate events
Chain	Send events to another pipeline
Code	Run custom JavaScript code on events
Tee	Copy events to a secondary destination
Auto Timestamp	Automatically detect and normalize timestamps
CEF Serializer	Serialize events to CEF format
Grok	Parse events using Grok patterns
Guard	Apply Cribl Guard (DLP) rules
Numerify	Convert string fields to numbers
SNMP Trap Serialize	Serialize SNMP trap events
OTLP Logs/Metrics/Traces	Convert to/from OpenTelemetry format
Event Breaker	Re-break events within a pipeline
Comment	Add documentation comments (no processing)

Function Configuration Pattern

Every function has:

• id: Function type (e.g., "eval", "mask", "drop")
• filter: JS expression determining which events this function processes (default: "true")
• disabled: Boolean toggle
• final: When true, matched events don't continue to next function
• conf: Function-specific configuration

---

Cribl Expressions (C.* helpers)

Use in any JavaScript expression field:

// Variables
C.vars.myVariable
C.vars.epochTime()          // expression-type variable

// Lookups
C.Lookup('lookup_file.csv', 'key_field')
C.Lookup('lookup_file.csv', 'key_field').match(source_ip, 'result_field')

// Masking
C.Mask.random(length)        // random alphanumeric
C.Mask.repeat(length, char)  // repeating character
C.Mask.md5(value)            // MD5 hash
C.Mask.sha1(value)           // SHA1 hash
C.Mask.sha256(value)         // SHA256 hash
C.Mask.REDACTED              // literal "REDACTED"

// Encryption
C.Crypto.encrypt(value, keyId, algorithm)
C.Crypto.decrypt(value, keyId)

// Encoding
C.Encode.uri(value)
C.Encode.base64(value)
C.Decode.base64(value)
C.Decode.uri(value)

// Secrets
C.Secret('secretName', 'type').value
C.Secret('api_key', 'keypair').apiKey

// Time
C.Time.now()
C.Time.strftime(epoch, format)
C.Time.strptime(string, format)

// Network
C.Net.cidrMatch(ip, cidr)

// Text
C.Text.hashCode(value)

---

Building an MCP Server for Cribl Stream

When building an MCP server, implement tools for each artifact type. Here is the recommended tool structure:

Required MCP Tools

// Authentication
cribl_login(host, username, password) → token

// Routes
cribl_list_routes(group?)
cribl_get_route(id, group?)
cribl_create_route(id, filter, pipeline, output, final?, group?)
cribl_update_route(id, updates, group?)
cribl_delete_route(id, group?)
cribl_reorder_routes(routeIds[], group?)

// Pipelines
cribl_list_pipelines(group?)
cribl_get_pipeline(id, group?)
cribl_create_pipeline(id, functions[], group?)
cribl_update_pipeline(id, updates, group?)
cribl_delete_pipeline(id, group?)
cribl_add_function(pipelineId, function, position?, group?)

// Sources
cribl_list_sources(group?)
cribl_get_source(id, group?)
cribl_create_source(id, type, config, group?)
cribl_update_source(id, updates, group?)
cribl_delete_source(id, group?)

// Destinations
cribl_list_destinations(group?)
cribl_get_destination(id, group?)
cribl_create_destination(id, type, config, group?)
cribl_update_destination(id, updates, group?)
cribl_delete_destination(id, group?)

// Knowledge Objects
cribl_list_variables(group?)
cribl_create_variable(id, type, value, description?, group?)
cribl_list_lookups(group?)
cribl_update_lookup(id, file_content, group?)
cribl_list_event_breakers(group?)

// Packs
cribl_list_packs(group?)
cribl_install_pack(source, group?)
cribl_export_pack(id, group?)

// Version Control
cribl_commit(message, group?)
cribl_deploy(group?)
cribl_sync()

// System
cribl_health()
cribl_system_info()
cribl_system_metrics()
cribl_restart(group?)
cribl_reload(group?)

// Workers (distributed)
cribl_list_groups()
cribl_list_workers(group?)

// Jobs
cribl_list_jobs(group?)
cribl_run_job(collectorId, config?, group?)
cribl_get_job_status(jobId, group?)

MCP Server Implementation Pattern (TypeScript)

import { McpServer } from "@anthropic-ai/mcp";

const CRIBL_HOST = process.env.CRIBL_HOST || "http://localhost:9000";
let authToken: string | null = null;

async function criblApi(method: string, path: string, body?: any) {
  const url = `${CRIBL_HOST}/api/v1${path}`;
  const res = await fetch(url, {
    method,
    headers: {
      "Content-Type": "application/json",
      ...(authToken ? { Authorization: `Bearer ${authToken}` } : {}),
    },
    body: body ? JSON.stringify(body) : undefined,
  });
  if (!res.ok) throw new Error(`Cribl API ${method} ${path}: ${res.status} ${await res.text()}`);
  return res.json();
}

async function login(username: string, password: string) {
  const result = await criblApi("POST", "/auth/login", { username, password });
  authToken = result.token;
  return result;
}

function groupPrefix(group?: string) {
  return group ? `/m/${group}` : "";
}

// Example tool implementations:

// List routes
async function listRoutes(group?: string) {
  return criblApi("GET", `${groupPrefix(group)}/routes`);
}

// Create a route
async function createRoute(route: {
  id: string;
  filter: string;
  pipeline: string;
  output: string;
  final?: boolean;
  description?: string;
}, group?: string) {
  return criblApi("POST", `${groupPrefix(group)}/routes`, {
    ...route,
    final: route.final ?? true,
  });
}

// Create a pipeline with functions
async function createPipeline(id: string, functions: any[], group?: string) {
  return criblApi("POST", `${groupPrefix(group)}/pipelines`, {
    id,
    conf: { functions },
  });
}

// Commit and deploy
async function commitAndDeploy(message: string, group?: string) {
  await criblApi("POST", `${groupPrefix(group)}/version/commit`, { message });
  await criblApi("POST", `${groupPrefix(group)}/version/deploy`);
}

Environment Variables

CRIBL_HOST=http://localhost:9000    # API host
CRIBL_USERNAME=admin               # Login username
CRIBL_PASSWORD=<password>          # Login password
CRIBL_API_PORT=9000                # API port (default 9000)

---

Common Source Types

Type	type value	Key Config
Syslog	syslog	host, port, protocol (tcp/udp)
Splunk HEC	splunk_hec	host, port, token
Splunk S2S	splunk	host, port
HTTP/S	http	host, port
Kafka	kafka	brokers[], topics[], groupId
Amazon S3	s3	bucket, region, queueName
Amazon SQS	sqs	queueName, region
Amazon Kinesis	kinesis	streamName, region
Azure Event Hubs	azure_eh	connectionString, consumerGroup
Google Pub/Sub	google_pubsub	projectId, subscriptionName
TCP JSON	tcp_json	host, port
Raw TCP	raw_tcp	host, port
Raw UDP	raw_udp	host, port
File Monitor	file	path, mode
Windows Events	windows_event_logs	channels[]
Elasticsearch API	elastic	url, index
Prometheus Remote Write	prometheus_rw	host, port
Grafana	grafana	host, port
Office 365	office365_activity	tenantId, clientId
REST Collector	collection	various
Script Collector	script	command, args

Common Destination Types

Type	type value	Key Config
Splunk HEC	splunk_hec	url, token
Splunk S2S	splunk	host, port
Splunk Load Balanced	splunk_lb	hosts[], port
Amazon S3	s3	bucket, region, stagePath
Elasticsearch	elastic	url, index
Kafka	kafka	brokers[], topic
Syslog	syslog	host, port, protocol, facility
Webhook	webhook	url, method, format
Azure Sentinel	sentinel	loginUrl, client_id
Google Cloud Storage	google_cloud_storage	bucket, region
Datadog	datadog	url, apiKey
New Relic	newrelic	url, licenseKey
Cribl HTTP	cribl_http	url
Cribl TCP	cribl_tcp	host, port
DevNull	devnull	(none)
Default	default	defaultId
Output Router	output_router	rules[]
Filesystem/NFS	filesystem	path
InfluxDB	influxdb	url, bucket, org, token
Prometheus	prometheus	url
Loki	loki	url
MinIO	minio	bucket, endpoint

---

Important Operational Notes

Changes That Require Reload (fast, no data loss)

• Functions, Pipelines, Packs, Routes
• Lookups, Parquet schemas, Global variables
• Group Settings > Limits, Logging levels

Changes That Require Restart (brief interruption)

• Sources, Destinations
• Event Breakers, QuickConnect configs
• Distributed mode changes, Worker Group assignment
• TLS settings, Worker process count

Route Evaluation Rules

1. Routes evaluate top-down in display order
2. Final=true (default): matched events stop, non-matched continue
3. Final=false: ALL events continue (matched events become clones)
4. Always use a catch-all route at the bottom (filter: "true")
5. Design for early reduction — broad filters first to minimize processing

Pipeline Best Practices

• Use function filters to narrow scope (source=='foo')
• Don't overload pipelines — separate pre/post-processing
• Use Eval for field manipulation, Parser for full parsing
• Use __e['field-name'] for fields with special characters

Filter Expression Examples

true                                        // match all
source=='syslog'                           // exact match
sourcetype=='access_combined' && status>=400 // compound
host.startsWith('web-')                    // string method
C.Net.cidrMatch(src_ip, '10.0.0.0/8')    // CIDR match
severity <= 3                              // numeric comparison
C.vars.blocked_hosts.includes(host)        // array lookup
/error|warn/i.test(_raw)                   // regex test
__e['user-agent'].includes('bot')          // special char fields
Math.random() < 0.1                        // 10% sampling

---

CLI Reference (Key Commands)

cribl start                    # Start Cribl Stream
cribl stop                     # Stop
cribl restart                  # Restart
cribl reload                   # Reload config (no restart)
cribl status                   # Show status
cribl version                  # Show version
cribl mode-single              # Switch to single-instance mode
cribl mode-master              # Switch to leader mode
cribl mode-worker              # Switch to worker mode

# Auth
cribl auth login -H http://host:9000 -u admin -p pass

# Variables
cribl vars add -i myVar -v 42 -t number -d "description"
cribl vars get
cribl vars update -i myVar -v 100
cribl vars remove -i myVar

# Pipelines
cat data.log | cribl pipe -p pipelineName

# Packs
cribl pack install <source>
cribl pack list
cribl pack export <packId>