Workers Observability migrate responses to TSV's (#140)

* migrate responses to TSV's

* formatting

* remove angry prompt

* tag wobs api usage as mcp

* pin deps

* format

Author: Ankcorn

apps/workers-observability/package.json

@@ -13,6 +13,7 @@
 	},
 	"dependencies": {
 		"@cloudflare/workers-oauth-provider": "0.0.5",
+		"@fast-csv/format": "5.0.2",
 		"@hono/zod-validator": "0.4.3",
 		"@modelcontextprotocol/sdk": "1.10.2",
 		"@repo/mcp-common": "workspace:*",

apps/workers-observability/src/tools/observability.ts

@@ -1,3 +1,5 @@
+import { writeToString } from '@fast-csv/format'
+
 import {
 	handleWorkerLogsKeys,
 	handleWorkerLogsValues,
@@ -60,6 +62,100 @@ This tool provides three primary views of your Worker data:
 			}
 			try {
 				const response = await queryWorkersObservability(agent.props.accessToken, accountId, query)
+
+				if (query.view === 'calculations') {
+					let data = ''
+					for (const calculation of response?.calculations || []) {
+						const alias = calculation.alias || calculation.calculation
+						const aggregates = calculation.aggregates.map((agg) => {
+							const keys = agg.groups?.reduce(
+								(acc, group) => {
+									acc[`${group.key}`] = `${group.value}`
+									return acc
+								},
+								{} as Record<string, string>
+							)
+							return {
+								...keys,
+								[alias]: agg.value,
+							}
+						})
+
+						const aggregatesString = await writeToString(aggregates, {
+							headers: true,
+							delimiter: '\t',
+						})
+
+						const series = calculation.series.map(({ time, data }) => {
+							return {
+								time,
+								...data.reduce(
+									(acc, point) => {
+										const key = point.groups?.reduce((acc, group) => {
+											return `${acc} * ${group.value}`
+										}, '')
+										if (!key) {
+											return {
+												...acc,
+												[alias]: point.value,
+											}
+										}
+										return {
+											...acc,
+											key,
+											[alias]: point.value,
+										}
+									},
+									{} as Record<string, string | number | undefined>
+								),
+							}
+						})
+						const seriesString = await writeToString(series, { headers: true, delimiter: '\t' })
+						data = data + '\n' + `## ${alias}`
+						data = data + '\n' + `### Aggregation`
+						data = data + '\n' + aggregatesString
+						data = data + '\n' + `### Series`
+						data = data + '\n' + seriesString
+					}
+
+					return {
+						content: [
+							{
+								type: 'text',
+								text: data,
+							},
+						],
+					}
+				}
+
+				if (query.view === 'events') {
+					const events = response?.events?.events
+					return {
+						content: [
+							{
+								type: 'text',
+								text: JSON.stringify(events),
+							},
+						],
+					}
+				}
+
+				if (query.view === 'invocations') {
+					const invocations = Object.entries(response?.invocations || {}).map(([_, logs]) => {
+						const invocationLog = logs.find((log) => log.$metadata.type === 'cf-worker-event')
+						return invocationLog?.$metadata ?? logs[0]?.$metadata
+					})
+
+					const tsv = await writeToString(invocations, { headers: true, delimiter: '\t' })
+					return {
+						content: [
+							{
+								type: 'text',
+								text: tsv,
+							},
+						],
+					}
+				}
 				return {
 					content: [
 						{
@@ -110,11 +206,16 @@ This tool provides three primary views of your Worker data:
 			}
 			try {
 				const result = await handleWorkerLogsKeys(agent.props.accessToken, accountId, keysQuery)
+
+				const tsv = await writeToString(
+					result.map((key) => ({ type: key.type, key: key.key })),
+					{ headers: true, delimiter: '\t' }
+				)
 				return {
 					content: [
 						{
 							type: 'text',
-							text: JSON.stringify(result),
+							text: tsv,
 						},
 					],
 				}
@@ -155,11 +256,15 @@ This tool provides three primary views of your Worker data:
 			}
 			try {
 				const result = await handleWorkerLogsValues(agent.props.accessToken, accountId, valuesQuery)
+				const tsv = await writeToString(
+					result?.map((value) => ({ type: value.type, value: value.value })) || [],
+					{ headers: true, delimiter: '\t' }
+				)
 				return {
 					content: [
 						{
 							type: 'text',
-							text: JSON.stringify(result),
+							text: tsv,
 						},
 					],
 				}

packages/mcp-common/src/api/workers-observability.ts

@@ -1,3 +1,5 @@
+import { env } from 'cloudflare:workers'
+
 import { fetchCloudflareApi } from '../cloudflare-api'
 import {
 	zKeysResponse,
@@ -23,6 +25,8 @@ export async function queryWorkersObservability(
 	accountId: string,
 	query: QueryRunRequest
 ): Promise<z.infer<typeof zReturnedQueryRunResult> | null> {
+	// @ts-expect-error We don't have actual env in this package
+	const environment = env.ENVIRONMENT
 	const data = await fetchCloudflareApi({
 		endpoint: '/workers/observability/telemetry/query',
 		accountId,
@@ -32,6 +36,7 @@ export async function queryWorkersObservability(
 			method: 'POST',
 			headers: {
 				'Content-Type': 'application/json',
+				'workers-observability-origin': `workers-observability-mcp-${environment}`,
 			},
 			body: JSON.stringify({ ...query, timeframe: fixTimeframe(query.timeframe) }),
 		},

packages/mcp-common/src/cloudflare-api.ts

@@ -40,6 +40,16 @@ export async function fetchCloudflareApi<T>({
 }): Promise<T> {
 	const url = `https://api.cloudflare.com/client/v4/accounts/${accountId}${endpoint}`
 
+	// @ts-expect-error We don't have actual env in this package
+	if (env.DEV_DISABLE_OAUTH) {
+		options.headers = {
+			...options.headers,
+			// @ts-expect-error We don't have actual env in this package
+			'X-Auth-Email': env.DEV_CLOUDFLARE_EMAIL,
+			// @ts-expect-error We don't have actual env in this package
+			'X-Auth-Key': env.DEV_CLOUDFLARE_API_TOKEN,
+		}
+	}
 	const response = await fetch(url, {
 		...options,
 		headers: {