feat(tinybird): added tinybird block (#2781)

2026-09-24 15:45:35 +08:00 · 2026-01-14 00:25:54 -08:00
parent bd7009e316
commit d5bd97de32
12 changed files with 644 additions and 0 deletions
@@ -1897,6 +1897,19 @@ export function TelegramIcon(props: SVGProps<SVGSVGElement>) {
  )
 }

+export function TinybirdIcon(props: SVGProps<SVGSVGElement>) {
+  return (
+    <svg {...props} xmlns='http://www.w3.org/2000/svg' viewBox='0 0 24 24' fill='none'>
+      <rect x='0' y='0' width='24' height='24' fill='#2EF598' rx='6' />
+      <g transform='translate(2, 2) scale(0.833)'>
+        <path d='M25 2.64 17.195.5 14.45 6.635z' fill='#1E7F63' />
+        <path d='M17.535 17.77 10.39 15.215 6.195 25.5z' fill='#1E7F63' />
+        <path d='M0 11.495 17.535 17.77 20.41 4.36z' fill='#1F2437' />
+      </g>
+    </svg>
+  )
+}
+
 export function ClayIcon(props: SVGProps<SVGSVGElement>) {
  return (
    <svg {...props} xmlns='http://www.w3.org/2000/svg' width='40' height='40' viewBox='0 0 400 400'>
@@ -107,6 +107,7 @@ import {
  SupabaseIcon,
  TavilyIcon,
  TelegramIcon,
+  TinybirdIcon,
  TranslateIcon,
  TrelloIcon,
  TTSIcon,
@@ -230,6 +231,8 @@ export const blockTypeToIconMap: Record<string, IconComponent> = {
  supabase: SupabaseIcon,
  tavily: TavilyIcon,
  telegram: TelegramIcon,
+  thinking: BrainIcon,
+  tinybird: TinybirdIcon,
  translate: TranslateIcon,
  trello: TrelloIcon,
  tts: TTSIcon,
@@ -103,6 +103,8 @@
    "supabase",
    "tavily",
    "telegram",
+    "thinking",
+    "tinybird",
    "translate",
    "trello",
    "tts",
@@ -0,0 +1,70 @@
+---
+title: Tinybird
+description: Send events and query data with Tinybird
+---
+
+import { BlockInfoCard } from "@/components/ui/block-info-card"
+
+<BlockInfoCard 
+  type="tinybird"
+  color="#2EF598"
+/>
+
+## Usage Instructions
+
+Interact with Tinybird using the Events API to stream JSON or NDJSON events, or use the Query API to execute SQL queries against Pipes and Data Sources.
+
+
+
+## Tools
+
+### `tinybird_events`
+
+Send events to a Tinybird Data Source using the Events API. Supports JSON and NDJSON formats with optional gzip compression.
+
+#### Input
+
+| Parameter | Type | Required | Description |
+| --------- | ---- | -------- | ----------- |
+| `base_url` | string | Yes | Tinybird API base URL \(e.g., https://api.tinybird.co or https://api.us-east.tinybird.co\) |
+| `datasource` | string | Yes | Name of the Tinybird Data Source to send events to |
+| `data` | string | Yes | Data to send as NDJSON \(newline-delimited JSON\) or JSON string. Each event should be a valid JSON object. |
+| `wait` | boolean | No | Wait for database acknowledgment before responding. Enables safer retries but introduces latency. Defaults to false. |
+| `format` | string | No | Format of the events data: "ndjson" \(default\) or "json" |
+| `compression` | string | No | Compression format: "none" \(default\) or "gzip" |
+| `token` | string | Yes | Tinybird API Token with DATASOURCE:APPEND or DATASOURCE:CREATE scope |
+
+#### Output
+
+| Parameter | Type | Description |
+| --------- | ---- | ----------- |
+| `successful_rows` | number | Number of rows successfully ingested |
+| `quarantined_rows` | number | Number of rows quarantined \(failed validation\) |
+
+### `tinybird_query`
+
+Execute SQL queries against Tinybird Pipes and Data Sources using the Query API.
+
+#### Input
+
+| Parameter | Type | Required | Description |
+| --------- | ---- | -------- | ----------- |
+| `base_url` | string | Yes | Tinybird API base URL \(e.g., https://api.tinybird.co\) |
+| `query` | string | Yes | SQL query to execute. Specify your desired output format \(e.g., FORMAT JSON, FORMAT CSV, FORMAT TSV\). JSON format provides structured data, while other formats return raw text. |
+| `pipeline` | string | No | Optional pipe name. When provided, enables SELECT * FROM _ syntax |
+| `token` | string | Yes | Tinybird API Token with PIPE:READ scope |
+
+#### Output
+
+| Parameter | Type | Description |
+| --------- | ---- | ----------- |
+| `data` | json | Query result data. For FORMAT JSON: array of objects. For other formats \(CSV, TSV, etc.\): raw text string. |
+| `rows` | number | Number of rows returned \(only available with FORMAT JSON\) |
+| `statistics` | json | Query execution statistics - elapsed time, rows read, bytes read \(only available with FORMAT JSON\) |
+
+
+
+## Notes
+
+- Category: `tools`
+- Type: `tinybird`
@@ -0,0 +1,207 @@
+import { TinybirdIcon } from '@/components/icons'
+import type { BlockConfig } from '@/blocks/types'
+import { AuthMode } from '@/blocks/types'
+import type { TinybirdResponse } from '@/tools/tinybird/types'
+
+export const TinybirdBlock: BlockConfig<TinybirdResponse> = {
+  type: 'tinybird',
+  name: 'Tinybird',
+  description: 'Send events and query data with Tinybird',
+  authMode: AuthMode.ApiKey,
+  longDescription:
+    'Interact with Tinybird using the Events API to stream JSON or NDJSON events, or use the Query API to execute SQL queries against Pipes and Data Sources.',
+  docsLink: 'https://www.tinybird.co/docs/api-reference',
+  category: 'tools',
+  bgColor: '#2EF598',
+  icon: TinybirdIcon,
+  subBlocks: [
+    {
+      id: 'operation',
+      title: 'Operation',
+      type: 'dropdown',
+      options: [
+        { label: 'Send Events', id: 'tinybird_events' },
+        { label: 'Query', id: 'tinybird_query' },
+      ],
+      value: () => 'tinybird_events',
+    },
+    {
+      id: 'base_url',
+      title: 'Base URL',
+      type: 'short-input',
+      placeholder: 'https://api.tinybird.co',
+      required: true,
+    },
+    {
+      id: 'token',
+      title: 'API Token',
+      type: 'short-input',
+      placeholder: 'Enter your Tinybird API token',
+      password: true,
+      required: true,
+    },
+    // Send Events operation inputs
+    {
+      id: 'datasource',
+      title: 'Data Source',
+      type: 'short-input',
+      placeholder: 'my_events_datasource',
+      condition: { field: 'operation', value: 'tinybird_events' },
+      required: true,
+    },
+    {
+      id: 'data',
+      title: 'Data',
+      type: 'code',
+      placeholder:
+        '{"event": "click", "timestamp": "2024-01-01T12:00:00Z"}\n{"event": "view", "timestamp": "2024-01-01T12:00:01Z"}',
+      condition: { field: 'operation', value: 'tinybird_events' },
+      required: true,
+    },
+    {
+      id: 'format',
+      title: 'Format',
+      type: 'dropdown',
+      options: [
+        { label: 'NDJSON (Newline-delimited JSON)', id: 'ndjson' },
+        { label: 'JSON', id: 'json' },
+      ],
+      value: () => 'ndjson',
+      condition: { field: 'operation', value: 'tinybird_events' },
+    },
+    {
+      id: 'compression',
+      title: 'Compression',
+      type: 'dropdown',
+      options: [
+        { label: 'None', id: 'none' },
+        { label: 'Gzip', id: 'gzip' },
+      ],
+      value: () => 'none',
+      mode: 'advanced',
+      condition: { field: 'operation', value: 'tinybird_events' },
+    },
+    {
+      id: 'wait',
+      title: 'Wait for Acknowledgment',
+      type: 'switch',
+      value: () => 'false',
+      mode: 'advanced',
+      condition: { field: 'operation', value: 'tinybird_events' },
+    },
+    // Query operation inputs
+    {
+      id: 'query',
+      title: 'SQL Query',
+      type: 'code',
+      placeholder: 'SELECT * FROM my_pipe FORMAT JSON\nOR\nSELECT * FROM my_pipe FORMAT CSV',
+      condition: { field: 'operation', value: 'tinybird_query' },
+      required: true,
+    },
+    {
+      id: 'pipeline',
+      title: 'Pipeline Name',
+      type: 'short-input',
+      placeholder: 'my_pipe (optional)',
+      condition: { field: 'operation', value: 'tinybird_query' },
+    },
+  ],
+  tools: {
+    access: ['tinybird_events', 'tinybird_query'],
+    config: {
+      tool: (params) => params.operation || 'tinybird_events',
+      params: (params) => {
+        const operation = params.operation || 'tinybird_events'
+        const result: Record<string, any> = {
+          base_url: params.base_url,
+          token: params.token,
+        }
+
+        if (operation === 'tinybird_events') {
+          // Send Events operation
+          if (!params.datasource) {
+            throw new Error('Data Source is required for Send Events operation')
+          }
+          if (!params.data) {
+            throw new Error('Data is required for Send Events operation')
+          }
+
+          result.datasource = params.datasource
+          result.data = params.data
+          result.format = params.format || 'ndjson'
+          result.compression = params.compression || 'none'
+
+          // Convert wait from string to boolean
+          // Convert wait from string to boolean
+          if (params.wait !== undefined) {
+            const waitValue =
+              typeof params.wait === 'string' ? params.wait.toLowerCase() : params.wait
+            result.wait = waitValue === 'true' || waitValue === true
+          }
+        } else if (operation === 'tinybird_query') {
+          // Query operation
+          if (!params.query) {
+            throw new Error('SQL Query is required for Query operation')
+          }
+
+          result.query = params.query
+          if (params.pipeline) {
+            result.pipeline = params.pipeline
+          }
+        }
+
+        return result
+      },
+    },
+  },
+  inputs: {
+    operation: { type: 'string', description: 'Operation to perform' },
+    base_url: { type: 'string', description: 'Tinybird API base URL' },
+    // Send Events inputs
+    datasource: {
+      type: 'string',
+      description: 'Name of the Tinybird Data Source',
+    },
+    data: {
+      type: 'string',
+      description: 'Data to send as JSON or NDJSON string',
+    },
+    wait: { type: 'boolean', description: 'Wait for database acknowledgment' },
+    format: {
+      type: 'string',
+      description: 'Format of the events (ndjson or json)',
+    },
+    compression: {
+      type: 'string',
+      description: 'Compression format (none or gzip)',
+    },
+    // Query inputs
+    query: { type: 'string', description: 'SQL query to execute' },
+    pipeline: { type: 'string', description: 'Optional pipeline name' },
+    // Common
+    token: { type: 'string', description: 'Tinybird API Token' },
+  },
+  outputs: {
+    // Send Events outputs
+    successful_rows: {
+      type: 'number',
+      description: 'Number of rows successfully ingested',
+    },
+    quarantined_rows: {
+      type: 'number',
+      description: 'Number of rows quarantined (failed validation)',
+    },
+    // Query outputs
+    data: {
+      type: 'json',
+      description:
+        'Query result data. FORMAT JSON: array of objects. Other formats (CSV, TSV, etc.): raw text string.',
+    },
+    rows: { type: 'number', description: 'Number of rows returned (only with FORMAT JSON)' },
+    statistics: {
+      type: 'json',
+      description:
+        'Query execution statistics - elapsed time, rows read, bytes read (only with FORMAT JSON)',
+    },
+  },
+}
@@ -121,6 +121,7 @@ import { SupabaseBlock } from '@/blocks/blocks/supabase'
 import { TavilyBlock } from '@/blocks/blocks/tavily'
 import { TelegramBlock } from '@/blocks/blocks/telegram'
 import { ThinkingBlock } from '@/blocks/blocks/thinking'
+import { TinybirdBlock } from '@/blocks/blocks/tinybird'
 import { TranslateBlock } from '@/blocks/blocks/translate'
 import { TrelloBlock } from '@/blocks/blocks/trello'
 import { TtsBlock } from '@/blocks/blocks/tts'
@@ -281,6 +282,7 @@ export const registry: Record<string, BlockConfig> = {
  tavily: TavilyBlock,
  telegram: TelegramBlock,
  thinking: ThinkingBlock,
+  tinybird: TinybirdBlock,
  translate: TranslateBlock,
  trello: TrelloBlock,
  twilio_sms: TwilioSMSBlock,
@@ -1897,6 +1897,19 @@ export function TelegramIcon(props: SVGProps<SVGSVGElement>) {
  )
 }

+export function TinybirdIcon(props: SVGProps<SVGSVGElement>) {
+  return (
+    <svg {...props} xmlns='http://www.w3.org/2000/svg' viewBox='0 0 24 24' fill='none'>
+      <rect x='0' y='0' width='24' height='24' fill='#2EF598' rx='6' />
+      <g transform='translate(2, 2) scale(0.833)'>
+        <path d='M25 2.64 17.195.5 14.45 6.635z' fill='#1E7F63' />
+        <path d='M17.535 17.77 10.39 15.215 6.195 25.5z' fill='#1E7F63' />
+        <path d='M0 11.495 17.535 17.77 20.41 4.36z' fill='#1F2437' />
+      </g>
+    </svg>
+  )
+}
+
 export function ClayIcon(props: SVGProps<SVGSVGElement>) {
  return (
    <svg {...props} xmlns='http://www.w3.org/2000/svg' width='40' height='40' viewBox='0 0 400 400'>
@@ -1381,6 +1381,7 @@ import {
  telegramSendVideoTool,
 } from '@/tools/telegram'
 import { thinkingTool } from '@/tools/thinking'
+import { tinybirdEventsTool, tinybirdQueryTool } from '@/tools/tinybird'
 import {
  trelloAddCommentTool,
  trelloCreateCardTool,
@@ -2237,6 +2238,8 @@ export const tools: Record<string, ToolConfig> = {
  apollo_email_accounts: apolloEmailAccountsTool,
  mistral_parser: mistralParserTool,
  thinking_tool: thinkingTool,
+  tinybird_events: tinybirdEventsTool,
+  tinybird_query: tinybirdQueryTool,
  stagehand_extract: stagehandExtractTool,
  stagehand_agent: stagehandAgentTool,
  mem0_add_memories: mem0AddMemoriesTool,
@@ -0,0 +1,128 @@
+import { gzipSync } from 'zlib'
+import { createLogger } from '@sim/logger'
+import type { TinybirdEventsParams, TinybirdEventsResponse } from '@/tools/tinybird/types'
+import type { ToolConfig } from '@/tools/types'
+
+const logger = createLogger('tinybird-events')
+
+export const eventsTool: ToolConfig<TinybirdEventsParams, TinybirdEventsResponse> = {
+  id: 'tinybird_events',
+  name: 'Tinybird Events',
+  description:
+    'Send events to a Tinybird Data Source using the Events API. Supports JSON and NDJSON formats with optional gzip compression.',
+  version: '1.0.0',
+  errorExtractor: 'nested-error-object',
+
+  params: {
+    base_url: {
+      type: 'string',
+      required: true,
+      visibility: 'user-only',
+      description:
+        'Tinybird API base URL (e.g., https://api.tinybird.co or https://api.us-east.tinybird.co)',
+    },
+    datasource: {
+      type: 'string',
+      required: true,
+      visibility: 'user-or-llm',
+      description: 'Name of the Tinybird Data Source to send events to',
+    },
+    data: {
+      type: 'string',
+      required: true,
+      visibility: 'user-or-llm',
+      description:
+        'Data to send as NDJSON (newline-delimited JSON) or JSON string. Each event should be a valid JSON object.',
+    },
+    wait: {
+      type: 'boolean',
+      required: false,
+      visibility: 'user-only',
+      description:
+        'Wait for database acknowledgment before responding. Enables safer retries but introduces latency. Defaults to false.',
+    },
+    format: {
+      type: 'string',
+      required: false,
+      visibility: 'user-only',
+      description: 'Format of the events data: "ndjson" (default) or "json"',
+    },
+    compression: {
+      type: 'string',
+      required: false,
+      visibility: 'user-only',
+      description: 'Compression format: "none" (default) or "gzip"',
+    },
+    token: {
+      type: 'string',
+      required: true,
+      visibility: 'user-only',
+      description: 'Tinybird API Token with DATASOURCE:APPEND or DATASOURCE:CREATE scope',
+    },
+  },
+
+  request: {
+    url: (params) => {
+      const baseUrl = params.base_url.endsWith('/') ? params.base_url.slice(0, -1) : params.base_url
+      const url = new URL(`${baseUrl}/v0/events`)
+      url.searchParams.set('name', params.datasource)
+      if (params.wait) {
+        url.searchParams.set('wait', 'true')
+      }
+      return url.toString()
+    },
+    method: 'POST',
+    headers: (params) => {
+      const headers: Record<string, string> = {
+        Authorization: `Bearer ${params.token}`,
+      }
+
+      if (params.compression === 'gzip') {
+        headers['Content-Encoding'] = 'gzip'
+      }
+
+      if (params.format === 'json') {
+        headers['Content-Type'] = 'application/json'
+      } else {
+        headers['Content-Type'] = 'application/x-ndjson'
+      }
+
+      return headers
+    },
+    body: (params) => {
+      const data = params.data
+      if (params.compression === 'gzip') {
+        return gzipSync(Buffer.from(data, 'utf-8'))
+      }
+      return data
+    },
+  },
+
+  transformResponse: async (response: Response) => {
+    const data = await response.json()
+
+    logger.info('Successfully sent events to Tinybird', {
+      successful: data.successful_rows,
+      quarantined: data.quarantined_rows,
+    })
+
+    return {
+      success: true,
+      output: {
+        successful_rows: data.successful_rows ?? 0,
+        quarantined_rows: data.quarantined_rows ?? 0,
+      },
+    }
+  },
+
+  outputs: {
+    successful_rows: {
+      type: 'number',
+      description: 'Number of rows successfully ingested',
+    },
+    quarantined_rows: {
+      type: 'number',
+      description: 'Number of rows quarantined (failed validation)',
+    },
+  },
+}
@@ -0,0 +1,5 @@
+import { eventsTool } from '@/tools/tinybird/events'
+import { queryTool } from '@/tools/tinybird/query'
+
+export const tinybirdEventsTool = eventsTool
+export const tinybirdQueryTool = queryTool
@@ -0,0 +1,139 @@
+import { createLogger } from '@sim/logger'
+import type { TinybirdQueryParams, TinybirdQueryResponse } from '@/tools/tinybird/types'
+import type { ToolConfig } from '@/tools/types'
+
+const logger = createLogger('tinybird-query')
+
+/**
+ * Tinybird Query Tool
+ *
+ * Executes SQL queries against Tinybird and returns results in the format specified in the query.
+ * - FORMAT JSON: Returns structured data with rows/statistics metadata
+ * - FORMAT CSV/TSV/etc: Returns raw text string
+ *
+ * The tool automatically detects the response format based on Content-Type headers.
+ */
+export const queryTool: ToolConfig<TinybirdQueryParams, TinybirdQueryResponse> = {
+  id: 'tinybird_query',
+  name: 'Tinybird Query',
+  description: 'Execute SQL queries against Tinybird Pipes and Data Sources using the Query API.',
+  version: '1.0.0',
+  errorExtractor: 'nested-error-object',
+
+  params: {
+    base_url: {
+      type: 'string',
+      required: true,
+      visibility: 'user-only',
+      description: 'Tinybird API base URL (e.g., https://api.tinybird.co)',
+    },
+    query: {
+      type: 'string',
+      required: true,
+      visibility: 'user-or-llm',
+      description:
+        'SQL query to execute. Specify your desired output format (e.g., FORMAT JSON, FORMAT CSV, FORMAT TSV). JSON format provides structured data, while other formats return raw text.',
+    },
+    pipeline: {
+      type: 'string',
+      required: false,
+      visibility: 'user-or-llm',
+      description: 'Optional pipe name. When provided, enables SELECT * FROM _ syntax',
+    },
+    token: {
+      type: 'string',
+      required: true,
+      visibility: 'user-only',
+      description: 'Tinybird API Token with PIPE:READ scope',
+    },
+  },
+
+  request: {
+    url: (params) => {
+      const baseUrl = params.base_url.endsWith('/') ? params.base_url.slice(0, -1) : params.base_url
+      return `${baseUrl}/v0/sql`
+    },
+    method: 'POST',
+    headers: (params) => ({
+      'Content-Type': 'application/x-www-form-urlencoded',
+      Authorization: `Bearer ${params.token}`,
+    }),
+    body: (params) => {
+      const searchParams = new URLSearchParams()
+      searchParams.set('q', params.query)
+      if (params.pipeline) {
+        searchParams.set('pipeline', params.pipeline)
+      }
+      return searchParams.toString()
+    },
+  },
+
+  transformResponse: async (response: Response) => {
+    const responseText = await response.text()
+    const contentType = response.headers.get('content-type') || ''
+
+    // Check if response is JSON based on content-type or try parsing
+    const isJson = contentType.includes('application/json') || contentType.includes('text/json')
+
+    if (isJson) {
+      try {
+        const data = JSON.parse(responseText)
+        logger.info('Successfully executed Tinybird query (JSON)', {
+          rows: data.rows,
+          elapsed: data.statistics?.elapsed,
+        })
+
+        return {
+          success: true,
+          output: {
+            data: data.data || [],
+            rows: data.rows || 0,
+            statistics: data.statistics
+              ? {
+                  elapsed: data.statistics.elapsed,
+                  rows_read: data.statistics.rows_read,
+                  bytes_read: data.statistics.bytes_read,
+                }
+              : undefined,
+          },
+        }
+      } catch (parseError) {
+        logger.error('Failed to parse JSON response', {
+          contentType,
+          parseError: parseError instanceof Error ? parseError.message : String(parseError),
+        })
+        throw new Error(
+          `Invalid JSON response: ${parseError instanceof Error ? parseError.message : 'Parse error'}`
+        )
+      }
+    }
+
+    // For non-JSON formats (CSV, TSV, etc.), return as raw text
+    logger.info('Successfully executed Tinybird query (non-JSON)', { contentType })
+    return {
+      success: true,
+      output: {
+        data: responseText,
+        rows: undefined,
+        statistics: undefined,
+      },
+    }
+  },
+
+  outputs: {
+    data: {
+      type: 'json',
+      description:
+        'Query result data. For FORMAT JSON: array of objects. For other formats (CSV, TSV, etc.): raw text string.',
+    },
+    rows: {
+      type: 'number',
+      description: 'Number of rows returned (only available with FORMAT JSON)',
+    },
+    statistics: {
+      type: 'json',
+      description:
+        'Query execution statistics - elapsed time, rows read, bytes read (only available with FORMAT JSON)',
+    },
+  },
+}
@@ -0,0 +1,59 @@
+import type { ToolResponse } from '@/tools/types'
+
+/**
+ * Base parameters for Tinybird API tools
+ */
+export interface TinybirdBaseParams {
+  token: string
+}
+
+/**
+ * Parameters for sending events to Tinybird
+ */
+export interface TinybirdEventsParams extends TinybirdBaseParams {
+  base_url: string
+  datasource: string
+  data: string
+  wait?: boolean
+  format?: 'ndjson' | 'json'
+  compression?: 'none' | 'gzip'
+}
+
+/**
+ * Response from sending events to Tinybird
+ */
+export interface TinybirdEventsResponse extends ToolResponse {
+  output: {
+    successful_rows: number
+    quarantined_rows: number
+  }
+}
+
+/**
+ * Parameters for querying Tinybird
+ */
+export interface TinybirdQueryParams extends TinybirdBaseParams {
+  base_url: string
+  query: string
+  pipeline?: string
+}
+
+/**
+ * Response from querying Tinybird
+ */
+export interface TinybirdQueryResponse extends ToolResponse {
+  output: {
+    data: unknown[] | string
+    rows?: number
+    statistics?: {
+      elapsed: number
+      rows_read: number
+      bytes_read: number
+    }
+  }
+}
+
+/**
+ * Union type for all possible Tinybird responses
+ */
+export type TinybirdResponse = TinybirdEventsResponse | TinybirdQueryResponse