diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index a5a57c7ed..cb8381d47 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -3,8 +3,9 @@ # @temporalio/sdk will be requested for review when # someone opens a pull request. * @temporalio/sdk -/ai-sdk/ @temporalio/sdk @temporalio/ai-sdk -/langsmith/ @temporalio/sdk @temporalio/ai-sdk -/openai-agents/ @temporalio/sdk @temporalio/ai-sdk -/strands-agents/ @temporalio/sdk @temporalio/ai-sdk -/workflow-streams/ @temporalio/sdk @temporalio/ai-sdk +/ai-sdk/ @temporalio/sdk @temporalio/ai-sdk +/google-adk-agents/ @temporalio/sdk @temporalio/ai-sdk +/langsmith/ @temporalio/sdk @temporalio/ai-sdk +/openai-agents/ @temporalio/sdk @temporalio/ai-sdk +/strands-agents/ @temporalio/sdk @temporalio/ai-sdk +/workflow-streams/ @temporalio/sdk @temporalio/ai-sdk diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e79c09b2..82c72fc4a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -60,6 +60,7 @@ jobs: eager-workflow-start early-return empty + google-adk-agents hello-world langsmith mutex diff --git a/.scripts/copy-shared-files.mjs b/.scripts/copy-shared-files.mjs index 8cb77b4b7..19fa5928a 100644 --- a/.scripts/copy-shared-files.mjs +++ b/.scripts/copy-shared-files.mjs @@ -62,6 +62,7 @@ const ESLINTIGNORE_EXCLUDE = [ const POST_CREATE_EXCLUDE = [ 'openai-agents', + 'google-adk-agents', 'env-config', 'dsl-interpreter', 'eager-workflow-start', diff --git a/.scripts/list-of-samples.json b/.scripts/list-of-samples.json index 56d850fe2..f0f769a6d 100644 --- a/.scripts/list-of-samples.json +++ b/.scripts/list-of-samples.json @@ -19,6 +19,7 @@ "expense", "fetch-esm", "food-delivery", + "google-adk-agents", "grpc-calls", "hello-world", "hello-world-js", diff --git a/README.md b/README.md index 34d6a84de..22cc6f879 100644 --- a/README.md +++ b/README.md @@ -181,6 +181,14 @@ and you'll be given the list of sample options. - [**Customer Service**](./openai-agents/customer-service): A long-running, multi-turn Workflow driven by Updates and Queries, with triage handoffs and `continueAsNew` to bound history. - [**Nexus Tools**](./openai-agents/nexus-tools): Expose a Nexus Operation as an agent tool with `nexusOperationAsTool`. - [**Streaming**](./openai-agents/src/streaming): Run an agent in streaming mode over a Workflow Stream, with an external client subscribing to the model's deltas live. +- [**Google ADK Agents**](./google-adk-agents): Run [Google Agent Development Kit](https://github.com/google/adk-js) (`@google/adk`) agents as Temporal Workflows with the `@temporalio/google-adk-agents` integration. The [`google-adk-agents/`](./google-adk-agents) directory contains seven samples: + - [**Basic**](./google-adk-agents/src/basic): A single `LlmAgent` whose model is a `TemporalModel`, driven by `InMemoryRunner` for one durable model call. + - [**Tools**](./google-adk-agents/src/tools): An existing Temporal Activity exposed to the agent as an ADK tool with `activityAsTool`. + - [**Agent Patterns**](./google-adk-agents/src/agent-patterns): A coordinator `LlmAgent` starts an ADK `transfer_to_agent` relay through a researcher and a writer, each with its own `TemporalModel`. + - [**MCP**](./google-adk-agents/src/mcp): A `TemporalMCPToolset` backed by a filesystem MCP server the Worker opens over stdio. + - [**Streaming**](./google-adk-agents/src/streaming): Token streaming from a direct `TemporalModel` call — no agent loop — over a Workflow Stream, with an external client printing the deltas as they arrive. + - [**Human Approval**](./google-adk-agents/src/human-approval): A `LongRunningFunctionTool` whose completion is gated by a Temporal Signal or Update. + - [**Observability**](./google-adk-agents/src/observability): Token usage, latency, and call counts from the agent loop's OpenTelemetry spans, by composing `OpenTelemetryPlugin` onto the Worker alongside `GoogleAdkPlugin`. ### Full-stack apps diff --git a/google-adk-agents/.eslintignore b/google-adk-agents/.eslintignore new file mode 100644 index 000000000..7bd99a41b --- /dev/null +++ b/google-adk-agents/.eslintignore @@ -0,0 +1,3 @@ +node_modules +lib +.eslintrc.js \ No newline at end of file diff --git a/google-adk-agents/.eslintrc.js b/google-adk-agents/.eslintrc.js new file mode 100644 index 000000000..9f199cd97 --- /dev/null +++ b/google-adk-agents/.eslintrc.js @@ -0,0 +1,48 @@ +const { builtinModules } = require('module'); + +const ALLOWED_NODE_BUILTINS = new Set(['assert']); + +module.exports = { + root: true, + parser: '@typescript-eslint/parser', + parserOptions: { + project: './tsconfig.json', + tsconfigRootDir: __dirname, + }, + plugins: ['@typescript-eslint', 'deprecation'], + extends: [ + 'eslint:recommended', + 'plugin:@typescript-eslint/eslint-recommended', + 'plugin:@typescript-eslint/recommended', + 'prettier', + ], + rules: { + // recommended for safety + '@typescript-eslint/no-floating-promises': 'error', // forgetting to await Activities and Workflow APIs is bad + 'deprecation/deprecation': 'warn', + + // code style preference + 'object-shorthand': ['error', 'always'], + + // relaxed rules, for convenience + '@typescript-eslint/no-unused-vars': [ + 'warn', + { + argsIgnorePattern: '^_', + varsIgnorePattern: '^_', + }, + ], + '@typescript-eslint/no-explicit-any': 'off', + }, + overrides: [ + { + files: ['src/**/workflows.ts', 'src/**/workflows-*.ts', 'src/**/workflows/*.ts'], + rules: { + 'no-restricted-imports': [ + 'error', + ...builtinModules.filter((m) => !ALLOWED_NODE_BUILTINS.has(m)).flatMap((m) => [m, `node:${m}`]), + ], + }, + }, + ], +}; diff --git a/google-adk-agents/.gitignore b/google-adk-agents/.gitignore new file mode 100644 index 000000000..a9f4ed545 --- /dev/null +++ b/google-adk-agents/.gitignore @@ -0,0 +1,2 @@ +lib +node_modules \ No newline at end of file diff --git a/google-adk-agents/.npmrc b/google-adk-agents/.npmrc new file mode 100644 index 000000000..9cf949503 --- /dev/null +++ b/google-adk-agents/.npmrc @@ -0,0 +1 @@ +package-lock=false \ No newline at end of file diff --git a/google-adk-agents/.nvmrc b/google-adk-agents/.nvmrc new file mode 100644 index 000000000..2bd5a0a98 --- /dev/null +++ b/google-adk-agents/.nvmrc @@ -0,0 +1 @@ +22 diff --git a/google-adk-agents/.post-create b/google-adk-agents/.post-create new file mode 100644 index 000000000..23ce8a44c --- /dev/null +++ b/google-adk-agents/.post-create @@ -0,0 +1,20 @@ +To begin development, install the Temporal CLI: + +Mac: {cyan brew install temporal} +Other: Download and extract the latest release from https://github.com/temporalio/cli/releases/latest + +Start Temporal Server: + +{cyan temporal server start-dev} + +Use Node version 22 or later: + +Mac: {cyan brew install node@22} +Other: https://nodejs.org/en/download/ + +This sample has several scenarios under {cyan src/}. Using two other shells, start a Worker for one scenario and run its client (example: {cyan basic}): + +{cyan GEMINI_API_KEY= npx ts-node src/basic/worker.ts} +{cyan npx ts-node src/basic/client.ts} + +See README.md for the full list of scenarios. diff --git a/google-adk-agents/.prettierignore b/google-adk-agents/.prettierignore new file mode 100644 index 000000000..7951405f8 --- /dev/null +++ b/google-adk-agents/.prettierignore @@ -0,0 +1 @@ +lib \ No newline at end of file diff --git a/google-adk-agents/.prettierrc b/google-adk-agents/.prettierrc new file mode 100644 index 000000000..965d50bff --- /dev/null +++ b/google-adk-agents/.prettierrc @@ -0,0 +1,2 @@ +printWidth: 120 +singleQuote: true diff --git a/google-adk-agents/README.md b/google-adk-agents/README.md new file mode 100644 index 000000000..380894322 --- /dev/null +++ b/google-adk-agents/README.md @@ -0,0 +1,28 @@ +# Google ADK Agents + +These samples use the `@temporalio/google-adk-agents` integration to run [Google Agent Development Kit](https://github.com/google/adk-js) (`@google/adk`) agents as durable Temporal Workflows. The ADK agent graph — the `Runner` loop, `LlmAgent`s, tools, and MCP toolsets — runs inside the Workflow and replays deterministically, while its non-deterministic I/O — model calls, MCP tool calls, and Activities exposed as tools — runs as durable Activities, so they retry on failure and are not repeated during Workflow replay. + +This is a single project: one `package.json` and one set of configs at the `google-adk-agents/` root, with each scenario in its own subdirectory under `src/`. Run `npm install` once here, then run any scenario by path (see each scenario's README). The integration package itself is documented in the [`@temporalio/google-adk-agents` README](https://github.com/temporalio/sdk-typescript/tree/main/contrib/google-adk-agents). + +## Prerequisites + +These apply to every sample in this directory: + +- A running Temporal dev server: `temporal server start-dev`. +- Node 22 or later. +- A Gemini API key for live runs: `export GEMINI_API_KEY=...`. +- Dependencies installed once at the `google-adk-agents/` root: `npm install`. + +Each scenario's README describes how to start its Worker and run its scenarios by path. + +## Samples + +| Sample | Demonstrates | +| :--------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------ | +| [`basic`](./src/basic) | A single `LlmAgent` whose model is a `TemporalModel`, driven by `InMemoryRunner` for one durable model call. | +| [`tools`](./src/tools) | An existing Temporal Activity exposed to the agent as an ADK tool via `activityAsTool`. | +| [`agent-patterns`](./src/agent-patterns) | A `transfer_to_agent` relay from a coordinator `LlmAgent` through a researcher and a writer, each with its own `TemporalModel`. | +| [`mcp`](./src/mcp) | A `TemporalMCPToolset` backed by an `mcpToolsets` factory on the plugin (a filesystem MCP server over stdio). | +| [`streaming`](./src/streaming) | Token streaming from a direct `TemporalModel` call — no agent loop — over the Workflow streams API. | +| [`human-approval`](./src/human-approval) | A `LongRunningFunctionTool` whose completion is gated by a Temporal Signal or Update. | +| [`observability`](./src/observability) | Token usage, latency, and call counts, by composing `OpenTelemetryPlugin` onto the Worker alongside `GoogleAdkPlugin`. | diff --git a/google-adk-agents/package.json b/google-adk-agents/package.json new file mode 100644 index 000000000..a73f0369a --- /dev/null +++ b/google-adk-agents/package.json @@ -0,0 +1,44 @@ +{ + "name": "temporal-google-adk-agents", + "version": "0.1.0", + "private": true, + "scripts": { + "build": "tsc --build", + "build.watch": "tsc --build --watch", + "format": "prettier --write .", + "format:check": "prettier --check .", + "lint": "eslint .", + "test": "mocha --exit --require ts-node/register --require source-map-support/register \"src/*/mocha/*.test.ts\"" + }, + "dependencies": { + "@temporalio/client": "^1.22.0", + "@temporalio/common": "^1.22.0", + "@temporalio/google-adk-agents": "^1.22.0", + "@temporalio/interceptors-opentelemetry": "^1.22.0", + "@temporalio/worker": "^1.22.0", + "@temporalio/workflow": "^1.22.0", + "@temporalio/workflow-streams": "^1.22.0", + "@google/adk": ">=1.5.0 <1.6.0", + "@google/genai": "^2.9.0", + "@opentelemetry/api": "^1.9.0", + "@opentelemetry/resources": "^1.25.1", + "@opentelemetry/sdk-trace-base": "^1.25.1", + "nanoid": "3.x" + }, + "devDependencies": { + "@temporalio/testing": "^1.22.0", + "@tsconfig/node22": "^22.0.0", + "@types/mocha": "8.x", + "@types/node": "^22.9.1", + "@typescript-eslint/eslint-plugin": "^8.18.0", + "@typescript-eslint/parser": "^8.18.0", + "eslint": "^8.57.1", + "eslint-config-prettier": "^9.1.0", + "eslint-plugin-deprecation": "^3.0.0", + "mocha": "8.x", + "prettier": "^3.4.2", + "ts-node": "^10.9.2", + "typescript": "^5.6.3", + "source-map-support": "^0.5.21" + } +} diff --git a/google-adk-agents/src/agent-patterns/README.md b/google-adk-agents/src/agent-patterns/README.md new file mode 100644 index 000000000..c5ca34220 --- /dev/null +++ b/google-adk-agents/src/agent-patterns/README.md @@ -0,0 +1,25 @@ +# Google ADK Agents: Agent Patterns + +A relay of three `LlmAgent`s over ADK's built-in `transfer_to_agent` tool, all of it running durably inside the Workflow: a coordinator transfers to a researcher, and the researcher transfers on to a writer, which produces the final answer. + +Each agent's `TemporalModel` sets a `summary` — the label the Temporal UI puts on that turn's `adk-invokeModel` Activity. + +## Run + +Run these from the `google-adk-agents/` root (run `npm install` there once first). + +```bash +# In one terminal, start the Worker (requires a local Temporal server and GEMINI_API_KEY): +GEMINI_API_KEY=... npx ts-node src/agent-patterns/worker.ts + +# In another terminal, run the scenario: +npx ts-node src/agent-patterns/client.ts +``` + +## Test + +```bash +npx mocha --exit --require ts-node/register --require source-map-support/register "src/agent-patterns/mocha/*.test.ts" +``` + +The test runs a real Worker against `TestWorkflowEnvironment` with a scripted `BaseLlm` double of its own, which answers each turn according to the agent ADK names as the asker: a transfer for the coordinator and the researcher, the haiku for the writer. No `GEMINI_API_KEY` is required. diff --git a/google-adk-agents/src/agent-patterns/client.ts b/google-adk-agents/src/agent-patterns/client.ts new file mode 100644 index 000000000..15ddd325c --- /dev/null +++ b/google-adk-agents/src/agent-patterns/client.ts @@ -0,0 +1,21 @@ +import { Connection, Client } from '@temporalio/client'; +import { nanoid } from 'nanoid'; +import { multiAgent } from './workflows'; + +async function run() { + const connection = await Connection.connect(); + const client = new Client({ connection }); + + const result = await client.workflow.execute(multiAgent, { + taskQueue: 'google-adk-agent-patterns', + workflowId: 'google-adk-agent-patterns-' + nanoid(), + args: ['durable execution'], + }); + + console.log(result); +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/agent-patterns/mocha/workflows.test.ts b/google-adk-agents/src/agent-patterns/mocha/workflows.test.ts new file mode 100644 index 000000000..6005940f8 --- /dev/null +++ b/google-adk-agents/src/agent-patterns/mocha/workflows.test.ts @@ -0,0 +1,81 @@ +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { BaseLlm } from '@google/adk'; +import type { BaseLlmConnection, LlmRequest, LlmResponse } from '@google/adk'; +import { after, before, describe, it } from 'mocha'; +import assert from 'assert'; +import { multiAgent } from '../workflows'; + +function text(s: string): LlmResponse { + return { content: { role: 'model', parts: [{ text: s }] }, turnComplete: true }; +} + +// ADK JS's `transfer_to_agent` tool reads `args.agentName` (camelCase). +function transferTo(agentName: string): LlmResponse { + return { + content: { role: 'model', parts: [{ functionCall: { name: 'transfer_to_agent', args: { agentName } } }] }, + turnComplete: true, + }; +} + +// Keyed by the asking agent rather than by call order, so an Activity retry re-serves the same turn. +function scriptedModelProvider(script: Record): (model: string) => BaseLlm { + class ScriptedLlm extends BaseLlm { + override async *generateContentAsync( + llmRequest: LlmRequest, + _stream = false, + _abortSignal?: AbortSignal, + ): AsyncGenerator { + const asking = llmRequest.config?.labels?.['adk_agent_name']; + const next = asking === undefined ? undefined : script[asking]; + if (next === undefined) { + throw new Error(`scripted model has no turn for agent '${asking}'`); + } + yield next; + } + + override async connect(_llmRequest: LlmRequest): Promise { + throw new Error('ScriptedLlm does not support connect().'); + } + } + return (model: string) => new ScriptedLlm({ model }); +} + +describe('google-adk-agents/agent-patterns workflow scenarios', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + it('multiAgent: the relay reaches the writer, and only the writer produces the final text', async () => { + const modelProvider = scriptedModelProvider({ + coordinator: transferTo('researcher'), + researcher: transferTo('writer'), + writer: text('snow on the mountain'), + }); + + const taskQueue = 'test-google-adk-agent-patterns'; + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + plugins: [new GoogleAdkPlugin({ modelProvider })], + }); + const result = await worker.runUntil( + testEnv.client.workflow.execute(multiAgent, { + args: ['mountains'], + workflowId: 'test-google-adk-agent-patterns-' + Date.now(), + taskQueue, + }), + ); + assert.strictEqual(result, 'snow on the mountain'); + }); +}); diff --git a/google-adk-agents/src/agent-patterns/worker.ts b/google-adk-agents/src/agent-patterns/worker.ts new file mode 100644 index 000000000..5edb64c26 --- /dev/null +++ b/google-adk-agents/src/agent-patterns/worker.ts @@ -0,0 +1,22 @@ +import { NativeConnection, Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; + +async function run() { + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + try { + const worker = await Worker.create({ + connection, + taskQueue: 'google-adk-agent-patterns', + workflowsPath: require.resolve('./workflows'), + plugins: [new GoogleAdkPlugin()], + }); + await worker.run(); + } finally { + await connection.close(); + } +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/agent-patterns/workflows.ts b/google-adk-agents/src/agent-patterns/workflows.ts new file mode 100644 index 000000000..b1bd5a5bb --- /dev/null +++ b/google-adk-agents/src/agent-patterns/workflows.ts @@ -0,0 +1,40 @@ +import { InMemoryRunner, LlmAgent, isFinalResponse, stringifyContent } from '@google/adk'; +import { TemporalModel } from '@temporalio/google-adk-agents/workflow'; + +export async function multiAgent(topic: string): Promise { + const researcher = new LlmAgent({ + name: 'researcher', + description: 'Reads a topic and hands it to whoever should write about it.', + model: new TemporalModel('gemini-2.5-flash', { summary: 'Researcher Agent' }), + instruction: 'You are a researcher. You write nothing yourself. Transfer the topic to the writer.', + }); + + const writer = new LlmAgent({ + name: 'writer', + description: 'Turns a topic into a haiku.', + model: new TemporalModel('gemini-2.5-flash', { summary: 'Writer Agent' }), + instruction: 'You are a poet. Write a haiku about the topic in the conversation.', + }); + + const coordinator = new LlmAgent({ + name: 'coordinator', + description: + 'Starts the relay by handing the incoming request to the first agent. Researches nothing and writes nothing.', + model: new TemporalModel('gemini-2.5-flash', { summary: 'Coordinator Agent' }), + instruction: 'You are a coordinator. Transfer the request to the researcher.', + subAgents: [researcher, writer], + }); + + const runner = new InMemoryRunner({ agent: coordinator }); + + let finalText = ''; + for await (const event of runner.runEphemeral({ + userId: 'user', + newMessage: { role: 'user', parts: [{ text: `Write a haiku about ${topic}.` }] }, + })) { + if (isFinalResponse(event)) { + finalText = stringifyContent(event); + } + } + return finalText; +} diff --git a/google-adk-agents/src/basic/README.md b/google-adk-agents/src/basic/README.md new file mode 100644 index 000000000..989108fe7 --- /dev/null +++ b/google-adk-agents/src/basic/README.md @@ -0,0 +1,23 @@ +# Google ADK Agents: Basic + +A single ADK `LlmAgent` whose model is a `TemporalModel`, driven by `InMemoryRunner` for one durable model call. Two things change from a vanilla ADK agent: the agent's `model` becomes `new TemporalModel('gemini-2.5-flash')` where the model name would otherwise go, and the Worker registers `GoogleAdkPlugin`. The agent loop then runs inside the Workflow while the model call runs as an Activity. + +## Run + +Run these from the `google-adk-agents/` root (run `npm install` there once first). + +```bash +# In one terminal, start the Worker (requires a local Temporal server and GEMINI_API_KEY): +GEMINI_API_KEY=... npx ts-node src/basic/worker.ts + +# In another terminal, run the scenario: +npx ts-node src/basic/client.ts +``` + +## Test + +```bash +npx mocha --exit --require ts-node/register --require source-map-support/register "src/basic/mocha/*.test.ts" +``` + +The tests run a real Worker against `TestWorkflowEnvironment`, so no `GEMINI_API_KEY` is required: one answers through `fakeModelProvider`, the other fails the model call and asserts the Workflow fails with it. diff --git a/google-adk-agents/src/basic/client.ts b/google-adk-agents/src/basic/client.ts new file mode 100644 index 000000000..c7454a340 --- /dev/null +++ b/google-adk-agents/src/basic/client.ts @@ -0,0 +1,21 @@ +import { Connection, Client } from '@temporalio/client'; +import { nanoid } from 'nanoid'; +import { helloWorld } from './workflows'; + +async function run() { + const connection = await Connection.connect(); + const client = new Client({ connection }); + + const result = await client.workflow.execute(helloWorld, { + taskQueue: 'google-adk-basic', + workflowId: 'google-adk-basic-' + nanoid(), + args: ['Write a haiku about durable execution.'], + }); + + console.log(result); +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/basic/mocha/workflows.test.ts b/google-adk-agents/src/basic/mocha/workflows.test.ts new file mode 100644 index 000000000..ed0890bf3 --- /dev/null +++ b/google-adk-agents/src/basic/mocha/workflows.test.ts @@ -0,0 +1,79 @@ +import { BaseLlm } from '@google/adk'; +import type { BaseLlmConnection, LlmResponse } from '@google/adk'; +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { ActivityFailure, ApplicationFailure } from '@temporalio/common'; +import { WorkflowFailedError } from '@temporalio/client'; +import { Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { fakeModelProvider } from '@temporalio/google-adk-agents/testing'; +import { after, before, describe, it } from 'mocha'; +import assert from 'assert'; +import { helloWorld } from '../workflows'; + +function failingModelProvider(): (model: string) => BaseLlm { + class FailingLlm extends BaseLlm { + override generateContentAsync(): AsyncGenerator { + throw ApplicationFailure.nonRetryable('the model rejected the request'); + } + + override async connect(): Promise { + throw new Error('FailingLlm does not support connect().'); + } + } + return (model: string) => new FailingLlm({ model }); +} + +describe('google-adk-agents/basic workflow scenarios', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + it('helloWorld: runs an LlmAgent through the runner with durable model calls', async () => { + const taskQueue = 'test-google-adk-basic'; + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + plugins: [new GoogleAdkPlugin({ modelProvider: fakeModelProvider() })], + }); + const result = await worker.runUntil( + testEnv.client.workflow.execute(helloWorld, { + args: ['Say hello.'], + workflowId: 'test-google-adk-basic-' + Date.now(), + taskQueue, + }), + ); + assert.strictEqual(result, 'fake-response:gemini-2.5-flash'); + }); + + it('helloWorld: a failed model call fails the Workflow rather than answering with an empty string', async () => { + const taskQueue = 'test-google-adk-basic-model-failure'; + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + plugins: [new GoogleAdkPlugin({ modelProvider: failingModelProvider() })], + }); + await worker.runUntil( + assert.rejects( + testEnv.client.workflow.execute(helloWorld, { + args: ['Say hello.'], + workflowId: taskQueue + '-' + Date.now(), + taskQueue, + }), + (err: unknown) => + err instanceof WorkflowFailedError && + err.cause instanceof ActivityFailure && + err.cause.cause?.message === 'the model rejected the request', + ), + ); + }); +}); diff --git a/google-adk-agents/src/basic/worker.ts b/google-adk-agents/src/basic/worker.ts new file mode 100644 index 000000000..20d242f27 --- /dev/null +++ b/google-adk-agents/src/basic/worker.ts @@ -0,0 +1,22 @@ +import { NativeConnection, Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; + +async function run() { + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + try { + const worker = await Worker.create({ + connection, + taskQueue: 'google-adk-basic', + workflowsPath: require.resolve('./workflows'), + plugins: [new GoogleAdkPlugin()], + }); + await worker.run(); + } finally { + await connection.close(); + } +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/basic/workflows.ts b/google-adk-agents/src/basic/workflows.ts new file mode 100644 index 000000000..f1a99e9c7 --- /dev/null +++ b/google-adk-agents/src/basic/workflows.ts @@ -0,0 +1,23 @@ +import { InMemoryRunner, LlmAgent, isFinalResponse, stringifyContent } from '@google/adk'; +import { TemporalModel } from '@temporalio/google-adk-agents/workflow'; + +export async function helloWorld(prompt: string): Promise { + const agent = new LlmAgent({ + name: 'assistant', + model: new TemporalModel('gemini-2.5-flash'), + instruction: 'You are a helpful assistant. Respond in a single sentence.', + }); + + const runner = new InMemoryRunner({ agent }); + + let finalText = ''; + for await (const event of runner.runEphemeral({ + userId: 'user', + newMessage: { role: 'user', parts: [{ text: prompt }] }, + })) { + if (isFinalResponse(event)) { + finalText = stringifyContent(event); + } + } + return finalText; +} diff --git a/google-adk-agents/src/human-approval/README.md b/google-adk-agents/src/human-approval/README.md new file mode 100644 index 000000000..83b3b029b --- /dev/null +++ b/google-adk-agents/src/human-approval/README.md @@ -0,0 +1,23 @@ +# Google ADK Agents: Human Approval + +A human-in-the-loop flow. The Workflow body invokes an ADK `LongRunningFunctionTool` whose `execute` blocks on a Temporal `condition` until a human's decision arrives, then returns it. The `approve` Signal and the `approveUpdate` Update each have their own handler; what they share is the variable the tool's `condition` is waiting on, so either one releases it (the Update handler additionally echoes the decision back to its caller). + +## Run + +Run these from the `google-adk-agents/` root (run `npm install` there once first). + +```bash +# In one terminal, start the Worker (requires a local Temporal server): +npx ts-node src/human-approval/worker.ts + +# In another terminal, start the Workflow (the client sends the approval Signal): +npx ts-node src/human-approval/client.ts +``` + +## Test + +```bash +npx mocha --exit --require ts-node/register --require source-map-support/register "src/human-approval/mocha/*.test.ts" +``` + +The tests run a real Worker against `TestWorkflowEnvironment`: two release the tool, one through the `approve` Signal and one through the `approveUpdate` Update, asserting it resumes with the supplied value; the third cancels the Workflow and asserts it ends CANCELLED. No `GEMINI_API_KEY` is required. diff --git a/google-adk-agents/src/human-approval/client.ts b/google-adk-agents/src/human-approval/client.ts new file mode 100644 index 000000000..411511c8b --- /dev/null +++ b/google-adk-agents/src/human-approval/client.ts @@ -0,0 +1,22 @@ +import { Connection, Client } from '@temporalio/client'; +import { nanoid } from 'nanoid'; +import { humanApproval, approveSignal } from './workflows'; + +async function run() { + const connection = await Connection.connect(); + const client = new Client({ connection }); + + const handle = await client.workflow.start(humanApproval, { + taskQueue: 'google-adk-human-approval', + workflowId: 'google-adk-human-approval-' + nanoid(), + }); + console.log(`Started workflow ${handle.workflowId}; sending approval Signal.`); + + await handle.signal(approveSignal, 'approved-by-operator'); + console.log(await handle.result()); +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/human-approval/mocha/workflows.test.ts b/google-adk-agents/src/human-approval/mocha/workflows.test.ts new file mode 100644 index 000000000..5d38ce934 --- /dev/null +++ b/google-adk-agents/src/human-approval/mocha/workflows.test.ts @@ -0,0 +1,70 @@ +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { after, before, describe, it } from 'mocha'; +import assert from 'assert'; +import { humanApproval, approveSignal, approveUpdate } from '../workflows'; + +describe('google-adk-agents/human-approval workflow scenarios', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + async function makeWorker(taskQueue: string) { + return Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + plugins: [new GoogleAdkPlugin()], + }); + } + + it('humanApproval: long-running tool resumes on the approve Signal', async () => { + const taskQueue = 'test-google-adk-human-approval-signal'; + const worker = await makeWorker(taskQueue); + await worker.runUntil(async () => { + const handle = await testEnv.client.workflow.start(humanApproval, { + workflowId: taskQueue + '-' + Date.now(), + taskQueue, + }); + await handle.signal(approveSignal, 'approved-via-signal'); + assert.strictEqual(await handle.result(), 'approved-via-signal'); + }); + }); + + it('humanApproval: long-running tool resumes on the approve Update', async () => { + const taskQueue = 'test-google-adk-human-approval-update'; + const worker = await makeWorker(taskQueue); + await worker.runUntil(async () => { + const handle = await testEnv.client.workflow.start(humanApproval, { + workflowId: taskQueue + '-' + Date.now(), + taskQueue, + }); + const updateResult = await handle.executeUpdate(approveUpdate, { args: ['approved-via-update'] }); + assert.strictEqual(updateResult, 'approved-via-update'); + assert.strictEqual(await handle.result(), 'approved-via-update'); + }); + }); + + it('humanApproval: cancelling the Workflow ends it as CANCELLED', async () => { + const taskQueue = 'test-google-adk-human-approval-cancel'; + const worker = await makeWorker(taskQueue); + await worker.runUntil(async () => { + const handle = await testEnv.client.workflow.start(humanApproval, { + workflowId: taskQueue + '-' + Date.now(), + taskQueue, + }); + await handle.cancel(); + await assert.rejects(handle.result()); + assert.strictEqual((await handle.describe()).status.name, 'CANCELLED'); + }); + }); +}); diff --git a/google-adk-agents/src/human-approval/worker.ts b/google-adk-agents/src/human-approval/worker.ts new file mode 100644 index 000000000..98e94d4f6 --- /dev/null +++ b/google-adk-agents/src/human-approval/worker.ts @@ -0,0 +1,22 @@ +import { NativeConnection, Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; + +async function run() { + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + try { + const worker = await Worker.create({ + connection, + taskQueue: 'google-adk-human-approval', + workflowsPath: require.resolve('./workflows'), + plugins: [new GoogleAdkPlugin()], + }); + await worker.run(); + } finally { + await connection.close(); + } +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/human-approval/workflows.ts b/google-adk-agents/src/human-approval/workflows.ts new file mode 100644 index 000000000..0acd2e588 --- /dev/null +++ b/google-adk-agents/src/human-approval/workflows.ts @@ -0,0 +1,43 @@ +import { LongRunningFunctionTool } from '@google/adk'; +import { + CancellationScope, + CancelledFailure, + condition, + defineSignal, + defineUpdate, + setHandler, +} from '@temporalio/workflow'; + +export const approveSignal = defineSignal<[string]>('approve'); +export const approveUpdate = defineUpdate('approveUpdate'); + +export async function humanApproval(): Promise { + let result: string | undefined; + + setHandler(approveSignal, (value) => { + result = value; + }); + setHandler(approveUpdate, (value) => { + result = value; + return value; + }); + + const tool = new LongRunningFunctionTool({ + name: 'humanApproval', + description: 'Wait for a human approval.', + execute: async () => { + await condition(() => result !== undefined); + return result; + }, + }); + + try { + return (await tool.runAsync({ args: {}, toolContext: {} as never })) as string; + } catch (err) { + // ADK's FunctionTool re-throws execute's error as a plain Error, so a cancelled condition() no longer reads as cancellation. + if (CancellationScope.current().consideredCancelled) { + throw new CancelledFailure('Workflow cancelled'); + } + throw err; + } +} diff --git a/google-adk-agents/src/mcp/README.md b/google-adk-agents/src/mcp/README.md new file mode 100644 index 000000000..70b1ad8c1 --- /dev/null +++ b/google-adk-agents/src/mcp/README.md @@ -0,0 +1,23 @@ +# Google ADK Agents: MCP + +A `TemporalMCPToolset` backed by a real [Model Context Protocol](https://modelcontextprotocol.io) server. The agent declares `new TemporalMCPToolset({ name: 'filesystem' })`; the Worker registers the matching `filesystem` factory on the plugin via `mcpToolsets`, which opens a filesystem MCP server over stdio (`@modelcontextprotocol/server-filesystem`). Tool discovery and every tool call route through `filesystem-listTools` / `filesystem-callTool` Activities, so the MCP connection details stay on the Worker and never enter Workflow inputs. + +## Run + +Run these from the `google-adk-agents/` root (run `npm install` there once first). The Worker spawns the filesystem MCP server with `npx`, exposing this sample's `src/mcp/sample-files/` directory. + +```bash +# In one terminal, start the Worker (requires a local Temporal server and GEMINI_API_KEY): +GEMINI_API_KEY=... npx ts-node src/mcp/worker.ts + +# In another terminal, run the scenario: +npx ts-node src/mcp/client.ts +``` + +## Test + +```bash +npx mocha --exit --require ts-node/register --require source-map-support/register "src/mcp/mocha/*.test.ts" +``` + +The test registers an in-memory `mockMCPToolset` on the plugin under the same `filesystem` name — no MCP server subprocess, no `npx`, no network, and no `GEMINI_API_KEY`. It drives `filesystemAgent` against a scripted model double, so a real tool call crosses the `filesystem-callTool` Activity and its result comes back on the next model turn. diff --git a/google-adk-agents/src/mcp/client.ts b/google-adk-agents/src/mcp/client.ts new file mode 100644 index 000000000..b30a8f033 --- /dev/null +++ b/google-adk-agents/src/mcp/client.ts @@ -0,0 +1,21 @@ +import { Connection, Client } from '@temporalio/client'; +import { nanoid } from 'nanoid'; +import { filesystemAgent } from './workflows'; + +async function run() { + const connection = await Connection.connect(); + const client = new Client({ connection }); + + const result = await client.workflow.execute(filesystemAgent, { + taskQueue: 'google-adk-mcp', + workflowId: 'google-adk-mcp-' + nanoid(), + args: ['List the files you can access, then summarize what is in hello.txt.'], + }); + + console.log(result); +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/mcp/mocha/workflows.test.ts b/google-adk-agents/src/mcp/mocha/workflows.test.ts new file mode 100644 index 000000000..2cf0d4266 --- /dev/null +++ b/google-adk-agents/src/mcp/mocha/workflows.test.ts @@ -0,0 +1,93 @@ +import { BaseLlm } from '@google/adk'; +import type { BaseLlmConnection, LlmRequest, LlmResponse } from '@google/adk'; +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { Worker } from '@temporalio/worker'; +import { Type } from '@google/genai'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { mockMCPToolset, type MockMCPToolDefinition } from '@temporalio/google-adk-agents/testing'; +import { after, before, describe, it } from 'mocha'; +import assert from 'assert'; +import { filesystemAgent } from '../workflows'; + +const readFileDef: MockMCPToolDefinition = { + declaration: { + name: 'read_file', + description: 'Read a file.', + parameters: { type: Type.OBJECT, properties: { path: { type: Type.STRING } }, required: ['path'] }, + }, + handler: (args) => ({ contents: `contents of ${String(args.path)}` }), +}; + +// A fresh model instance per Activity invocation, so the turn has to come from the request. +function readFileModelProvider(): (model: string) => BaseLlm { + class ReadFileLlm extends BaseLlm { + override async *generateContentAsync( + llmRequest: LlmRequest, + _stream = false, + _abortSignal?: AbortSignal, + ): AsyncGenerator { + const toolResponse = (llmRequest.contents ?? []) + .flatMap((content) => content.parts ?? []) + .find((part) => part.functionResponse?.name === 'read_file')?.functionResponse?.response; + if (toolResponse === undefined) { + yield { + content: { + role: 'model', + parts: [{ functionCall: { name: 'read_file', args: { path: 'hello.txt' } } }], + }, + turnComplete: true, + }; + return; + } + const text = String((toolResponse as { contents?: unknown }).contents); + yield { content: { role: 'model', parts: [{ text }] }, turnComplete: true }; + } + + override async connect(_llmRequest: LlmRequest): Promise { + throw new Error('ReadFileLlm does not support connect().'); + } + } + return (model: string) => new ReadFileLlm({ model }); +} + +describe('google-adk-agents/mcp workflow scenarios', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + it('filesystemAgent: the model calls an MCP tool and its result reaches the next turn', async () => { + const taskQueue = 'test-google-adk-mcp-agent'; + const workflowId = taskQueue + '-' + Date.now(); + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + plugins: [ + new GoogleAdkPlugin({ + modelProvider: readFileModelProvider(), + mcpToolsets: { filesystem: mockMCPToolset([readFileDef]) }, + }), + ], + }); + const result = await worker.runUntil( + testEnv.client.workflow.execute(filesystemAgent, { + args: ['Summarize hello.txt.'], + workflowId, + taskQueue, + }), + ); + assert.strictEqual(result, 'contents of hello.txt'); + + const { events } = await testEnv.client.workflow.getHandle(workflowId).fetchHistory(); + const scheduled = (events ?? []).map((e) => e.activityTaskScheduledEventAttributes?.activityType?.name); + assert.strictEqual(scheduled.filter((name) => name === 'filesystem-callTool').length, 1); + }); +}); diff --git a/google-adk-agents/src/mcp/sample-files/hello.txt b/google-adk-agents/src/mcp/sample-files/hello.txt new file mode 100644 index 000000000..18f9f10c2 --- /dev/null +++ b/google-adk-agents/src/mcp/sample-files/hello.txt @@ -0,0 +1,2 @@ +Hello from the Temporal Google ADK filesystem MCP sample. +This file is exposed to the agent through the filesystem MCP server. diff --git a/google-adk-agents/src/mcp/worker.ts b/google-adk-agents/src/mcp/worker.ts new file mode 100644 index 000000000..dba6cd711 --- /dev/null +++ b/google-adk-agents/src/mcp/worker.ts @@ -0,0 +1,37 @@ +import * as path from 'path'; +import { NativeConnection, Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; + +async function run() { + const exposedDir = path.resolve(__dirname, 'sample-files'); + + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + try { + const worker = await Worker.create({ + connection, + taskQueue: 'google-adk-mcp', + workflowsPath: require.resolve('./workflows'), + plugins: [ + new GoogleAdkPlugin({ + mcpToolsets: { + filesystem: () => ({ + type: 'StdioConnectionParams', + serverParams: { + command: 'npx', + args: ['-y', '@modelcontextprotocol/server-filesystem', exposedDir], + }, + }), + }, + }), + ], + }); + await worker.run(); + } finally { + await connection.close(); + } +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/mcp/workflows.ts b/google-adk-agents/src/mcp/workflows.ts new file mode 100644 index 000000000..e8de68e70 --- /dev/null +++ b/google-adk-agents/src/mcp/workflows.ts @@ -0,0 +1,24 @@ +import { InMemoryRunner, LlmAgent, isFinalResponse, stringifyContent } from '@google/adk'; +import { TemporalMCPToolset, TemporalModel } from '@temporalio/google-adk-agents/workflow'; + +export async function filesystemAgent(prompt: string): Promise { + const agent = new LlmAgent({ + name: 'filesystem_agent', + model: new TemporalModel('gemini-2.5-flash'), + instruction: 'Use your tools to answer questions about files.', + tools: [new TemporalMCPToolset({ name: 'filesystem' })], + }); + + const runner = new InMemoryRunner({ agent }); + + let finalText = ''; + for await (const event of runner.runEphemeral({ + userId: 'user', + newMessage: { role: 'user', parts: [{ text: prompt }] }, + })) { + if (isFinalResponse(event)) { + finalText = stringifyContent(event); + } + } + return finalText; +} diff --git a/google-adk-agents/src/observability/README.md b/google-adk-agents/src/observability/README.md new file mode 100644 index 000000000..472c83fb3 --- /dev/null +++ b/google-adk-agents/src/observability/README.md @@ -0,0 +1,37 @@ +# Google ADK Agents: Observability + +Where an agent's token usage, latency, and call counts come from once `OpenTelemetryPlugin` is composed onto the Worker alongside `GoogleAdkPlugin`, as this sample's `worker.ts` does. Without it, ADK's agent-loop spans are created inside the Workflow sandbox and dropped; the [`@temporalio/google-adk-agents` README](https://github.com/temporalio/sdk-typescript/tree/main/contrib/google-adk-agents#telemetry-and-observability) covers why, and the caveats that come with exporting them. + +Everything here is traces. ADK defines no OpenTelemetry metric instruments, so there is no metric stream to scrape — the numbers below are span attributes. + +## What the spans carry + +`call_llm`, one per model call: + +- `gen_ai.usage.input_tokens` and `gen_ai.usage.output_tokens`. There is no total; add the two, as this sample's Worker does when it prints each call. +- `gen_ai.request.model`. +- Latency is the span's own duration. ADK records no latency attribute, on this span or any other. + +`invoke_agent ` carries nothing numeric, so a call count means counting `call_llm` spans. + +## Run + +Run these from the `google-adk-agents/` root (run `npm install` there once first). + +```bash +# In one terminal, start the Worker (requires a local Temporal server and GEMINI_API_KEY): +GEMINI_API_KEY=... npx ts-node src/observability/worker.ts + +# In another terminal, run the scenario: +npx ts-node src/observability/client.ts +``` + +The per-call token, model, and latency lines print in the Worker's terminal. + +## Test + +```bash +npx mocha --exit --require ts-node/register --require source-map-support/register "src/observability/mocha/*.test.ts" +``` + +The test runs a real Worker against `TestWorkflowEnvironment` with `fakeModelProvider`, asserting that ADK's `call_llm` spans reach the span processor carrying their model name and token counts. No `GEMINI_API_KEY` is required. diff --git a/google-adk-agents/src/observability/adk-usage-span-processor.ts b/google-adk-agents/src/observability/adk-usage-span-processor.ts new file mode 100644 index 000000000..7747daffa --- /dev/null +++ b/google-adk-agents/src/observability/adk-usage-span-processor.ts @@ -0,0 +1,33 @@ +import type { Context } from '@opentelemetry/api'; +import type { ReadableSpan, Span, SpanProcessor } from '@opentelemetry/sdk-trace-base'; + +// ADK's tracer name; the SDK's own interceptor spans arrive under a different one. +const ADK_TRACER = 'gcp.vertex.agent'; + +export interface ModelCall { + model: string; + inputTokens: number; + outputTokens: number; + durationMs: number; +} + +export class AdkUsageSpanProcessor implements SpanProcessor { + constructor(private readonly onModelCall: (call: ModelCall) => void) {} + + onStart(_span: Span, _parentContext: Context): void {} + + onEnd(span: ReadableSpan): void { + if (span.instrumentationLibrary.name !== ADK_TRACER || span.name !== 'call_llm') return; + + this.onModelCall({ + model: String(span.attributes['gen_ai.request.model'] ?? ''), + inputTokens: Number(span.attributes['gen_ai.usage.input_tokens'] ?? 0), + outputTokens: Number(span.attributes['gen_ai.usage.output_tokens'] ?? 0), + durationMs: span.duration[0] * 1e3 + span.duration[1] / 1e6, + }); + } + + async forceFlush(): Promise {} + + async shutdown(): Promise {} +} diff --git a/google-adk-agents/src/observability/client.ts b/google-adk-agents/src/observability/client.ts new file mode 100644 index 000000000..4f69002e6 --- /dev/null +++ b/google-adk-agents/src/observability/client.ts @@ -0,0 +1,21 @@ +import { Connection, Client } from '@temporalio/client'; +import { nanoid } from 'nanoid'; +import { observedAgent } from './workflows'; + +async function run() { + const connection = await Connection.connect(); + const client = new Client({ connection }); + + const result = await client.workflow.execute(observedAgent, { + taskQueue: 'google-adk-observability', + workflowId: 'google-adk-observability-' + nanoid(), + args: [['Write a haiku about durable execution.', 'Write a haiku about workflow replay.']], + }); + + console.log(result.join('\n')); +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/observability/mocha/workflows.test.ts b/google-adk-agents/src/observability/mocha/workflows.test.ts new file mode 100644 index 000000000..854bb87d0 --- /dev/null +++ b/google-adk-agents/src/observability/mocha/workflows.test.ts @@ -0,0 +1,72 @@ +import type { LlmResponse } from '@google/adk'; +import { Resource } from '@opentelemetry/resources'; +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { fakeModelProvider } from '@temporalio/google-adk-agents/testing'; +import { OpenTelemetryPlugin } from '@temporalio/interceptors-opentelemetry'; +import { after, before, describe, it } from 'mocha'; +import assert from 'assert'; +import { AdkUsageSpanProcessor, type ModelCall } from '../adk-usage-span-processor'; +import { observedAgent } from '../workflows'; + +const prompts = ['Write a haiku about durable execution.', 'Write a haiku about workflow replay.']; + +const responses: LlmResponse[] = [ + { + content: { role: 'model', parts: [{ text: 'snow on the mountain' }] }, + usageMetadata: { promptTokenCount: 11, candidatesTokenCount: 7 }, + turnComplete: true, + }, +]; + +describe('google-adk-agents/observability workflow scenarios', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + it('observedAgent: composing OpenTelemetryPlugin exports the ADK spans and their token usage', async () => { + const taskQueue = 'test-google-adk-observability'; + const modelCalls: ModelCall[] = []; + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + plugins: [ + new OpenTelemetryPlugin({ + resource: new Resource({ 'service.name': 'test-google-adk-observability' }), + spanProcessor: new AdkUsageSpanProcessor((call) => modelCalls.push(call)), + }), + new GoogleAdkPlugin({ modelProvider: fakeModelProvider(responses) }), + ], + }); + const result = await worker.runUntil( + testEnv.client.workflow.execute(observedAgent, { + args: [prompts], + workflowId: `${taskQueue}-${Date.now()}`, + taskQueue, + }), + ); + assert.deepStrictEqual(result, ['snow on the mountain', 'snow on the mountain']); + + assert.ok( + modelCalls.length >= prompts.length, + `expected at least ${prompts.length} call_llm spans, got ${modelCalls.length}`, + ); + for (const call of modelCalls) { + assert.deepStrictEqual( + { model: call.model, inputTokens: call.inputTokens, outputTokens: call.outputTokens }, + { model: 'gemini-2.5-flash', inputTokens: 11, outputTokens: 7 }, + ); + assert.ok(call.durationMs > 0, 'each call_llm span should have a duration'); + } + }); +}); diff --git a/google-adk-agents/src/observability/worker.ts b/google-adk-agents/src/observability/worker.ts new file mode 100644 index 000000000..451b08ace --- /dev/null +++ b/google-adk-agents/src/observability/worker.ts @@ -0,0 +1,38 @@ +import { Resource } from '@opentelemetry/resources'; +import { NativeConnection, Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { OpenTelemetryPlugin } from '@temporalio/interceptors-opentelemetry'; +import { AdkUsageSpanProcessor } from './adk-usage-span-processor'; + +async function run() { + const spanProcessor = new AdkUsageSpanProcessor((call) => + console.log( + `call_llm ${call.model}: ${call.inputTokens} in + ${call.outputTokens} out = ` + + `${call.inputTokens + call.outputTokens} tokens in ${call.durationMs.toFixed(0)}ms`, + ), + ); + + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + try { + const worker = await Worker.create({ + connection, + taskQueue: 'google-adk-observability', + workflowsPath: require.resolve('./workflows'), + plugins: [ + new OpenTelemetryPlugin({ + resource: new Resource({ 'service.name': 'google-adk-observability' }), + spanProcessor, + }), + new GoogleAdkPlugin(), + ], + }); + await worker.run(); + } finally { + await connection.close(); + } +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/observability/workflows.ts b/google-adk-agents/src/observability/workflows.ts new file mode 100644 index 000000000..34a7f8227 --- /dev/null +++ b/google-adk-agents/src/observability/workflows.ts @@ -0,0 +1,27 @@ +import { InMemoryRunner, LlmAgent, isFinalResponse, stringifyContent } from '@google/adk'; +import { TemporalModel } from '@temporalio/google-adk-agents/workflow'; + +export async function observedAgent(prompts: string[]): Promise { + const agent = new LlmAgent({ + name: 'assistant', + model: new TemporalModel('gemini-2.5-flash'), + instruction: 'You are a helpful assistant. Respond in a single sentence.', + }); + + const runner = new InMemoryRunner({ agent }); + + const answers: string[] = []; + for (const prompt of prompts) { + let finalText = ''; + for await (const event of runner.runEphemeral({ + userId: 'user', + newMessage: { role: 'user', parts: [{ text: prompt }] }, + })) { + if (isFinalResponse(event)) { + finalText = stringifyContent(event); + } + } + answers.push(finalText); + } + return answers; +} diff --git a/google-adk-agents/src/streaming/README.md b/google-adk-agents/src/streaming/README.md new file mode 100644 index 000000000..8e9efd7e0 --- /dev/null +++ b/google-adk-agents/src/streaming/README.md @@ -0,0 +1,29 @@ +# Google ADK Agents: Streaming + +Token streaming over a Workflow Stream. Unlike the agent scenarios, the Workflow here calls a `TemporalModel` directly — no `LlmAgent`, no `Runner` — so the streaming path stands on its own. + +Streaming is requested by `generateContentAsync`'s `stream` argument, not by configuration; a streaming call additionally needs a `streamingTopic` on the `TemporalModel` to publish to, and fails without one. Together they route the call to an `adk-invokeModelStreaming` Activity, which publishes every `LlmResponse` the model yields to that topic via the Workflow streams API while still returning the same ordered sequence to the Workflow — the deterministic, replay-safe channel. Deltas arrive as `partial` responses and the turn ends with one non-partial response carrying its whole text; the Workflow returns that text and a count of the deltas. + +The Workflow hosts the [Workflow Stream](https://github.com/temporalio/sdk-typescript/tree/main/contrib/workflow-streams) with `new WorkflowStream()` and, once the model turn is done, waits up to 10 seconds for a `consumer-done` Signal before returning — completing would discard the log out from under a subscriber's final poll, and the bound keeps the Workflow from hanging when nobody is subscribed. The client subscribes to the topic from outside with `WorkflowStreamClient.create(client, workflowId).topic(streamingTopic).subscribe()`, prints each delta's text as it arrives, and stops at the non-partial response, at which point it Signals the Workflow. + +## Run + +Run these from the `google-adk-agents/` root (run `npm install` there once first). + +```bash +# In one terminal, start the Worker (requires a local Temporal server and GEMINI_API_KEY): +GEMINI_API_KEY=... npx ts-node src/streaming/worker.ts + +# In another terminal, run the scenario: +npx ts-node src/streaming/client.ts +``` + +The story prints token by token as the model produces it. In the Temporal UI the history shows a single `adk-invokeModelStreaming` Activity. + +## Test + +```bash +npx mocha --exit --require ts-node/register --require source-map-support/register "src/streaming/mocha/*.test.ts" +``` + +The test runs a real Worker against `TestWorkflowEnvironment` with a `BaseLlm` double scripted as a real text turn — three `partial` deltas then the non-partial whole-turn response. The double holds each response back until an external subscriber has taken the one before it, so the deltas reach the subscriber while the model call is still in flight. It asserts the subscriber receives all four in order, and that the Workflow returns the turn's text once and a count of three chunks. No `GEMINI_API_KEY` is required. diff --git a/google-adk-agents/src/streaming/client.ts b/google-adk-agents/src/streaming/client.ts new file mode 100644 index 000000000..64486542f --- /dev/null +++ b/google-adk-agents/src/streaming/client.ts @@ -0,0 +1,35 @@ +import type { LlmResponse } from '@google/adk'; +import { Connection, Client } from '@temporalio/client'; +import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; +import { nanoid } from 'nanoid'; +import { consumerDoneSignal, streamingModelCall, streamingTopic } from './workflows'; + +async function run() { + const connection = await Connection.connect(); + const client = new Client({ connection }); + + const workflowId = 'google-adk-streaming-' + nanoid(); + const handle = await client.workflow.start(streamingModelCall, { + taskQueue: 'google-adk-streaming', + workflowId, + args: ['Tell me a short story about a robot learning to paint.'], + }); + console.log(`Started workflow ${handle.workflowId}`); + + const streamClient = WorkflowStreamClient.create(client, workflowId); + for await (const item of streamClient.topic(streamingTopic).subscribe()) { + if (item.data.partial !== true) break; + for (const part of item.data.content?.parts ?? []) { + if (part.text) process.stdout.write(part.text); + } + } + await handle.signal(consumerDoneSignal); + + const result = await handle.result(); + console.log(`\n---\nReceived ${result.chunks} chunk(s).`); +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/streaming/mocha/workflows.test.ts b/google-adk-agents/src/streaming/mocha/workflows.test.ts new file mode 100644 index 000000000..a9dd2b3a6 --- /dev/null +++ b/google-adk-agents/src/streaming/mocha/workflows.test.ts @@ -0,0 +1,88 @@ +import { BaseLlm } from '@google/adk'; +import type { BaseLlmConnection, LlmResponse } from '@google/adk'; +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; +import { after, before, describe, it } from 'mocha'; +import assert from 'assert'; +import { consumerDoneSignal, streamingModelCall, streamingTopic } from '../workflows'; + +const turn: LlmResponse[] = [ + { content: { role: 'model', parts: [{ text: 'Hello ' }] }, partial: true }, + { content: { role: 'model', parts: [{ text: 'streaming ' }] }, partial: true }, + { content: { role: 'model', parts: [{ text: 'world' }] }, partial: true }, + { content: { role: 'model', parts: [{ text: 'Hello streaming world' }] }, partial: false, turnComplete: true }, +]; + +function gatedModelProvider(taken: () => number): (model: string) => BaseLlm { + class GatedLlm extends BaseLlm { + override async *generateContentAsync(): AsyncGenerator { + for (const [index, response] of turn.entries()) { + while (taken() < index) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + yield response; + } + } + + override async connect(): Promise { + throw new Error('GatedLlm does not support connect().'); + } + } + return (model: string) => new GatedLlm({ model }); +} + +describe('google-adk-agents/streaming workflow scenarios', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + it('streamingModelCall: an external subscriber receives the chunks while the call is in flight', async () => { + const taskQueue = 'test-google-adk-streaming'; + const received: LlmResponse[] = []; + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + plugins: [new GoogleAdkPlugin({ modelProvider: gatedModelProvider(() => received.length) })], + }); + + const workflowId = 'test-google-adk-streaming-' + Date.now(); + const result = await worker.runUntil(async () => { + const handle = await testEnv.client.workflow.start(streamingModelCall, { + args: ['stream please'], + workflowId, + taskQueue, + }); + + const streamClient = WorkflowStreamClient.create(testEnv.client, workflowId); + const gen = streamClient.topic(streamingTopic).subscribe(0, { pollCooldown: 0 }); + for await (const item of gen) { + received.push(item.data); + if (received.length >= turn.length) { + await gen.return(); + break; + } + } + await handle.signal(consumerDoneSignal); + + assert.deepStrictEqual( + received.map((response) => response.content?.parts?.[0]?.text), + ['Hello ', 'streaming ', 'world', 'Hello streaming world'], + ); + return handle.result(); + }); + + assert.strictEqual(result.text, 'Hello streaming world'); + assert.strictEqual(result.chunks, 3); + }); +}); diff --git a/google-adk-agents/src/streaming/worker.ts b/google-adk-agents/src/streaming/worker.ts new file mode 100644 index 000000000..5f8f0df11 --- /dev/null +++ b/google-adk-agents/src/streaming/worker.ts @@ -0,0 +1,22 @@ +import { NativeConnection, Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; + +async function run() { + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + try { + const worker = await Worker.create({ + connection, + taskQueue: 'google-adk-streaming', + workflowsPath: require.resolve('./workflows'), + plugins: [new GoogleAdkPlugin()], + }); + await worker.run(); + } finally { + await connection.close(); + } +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/streaming/workflows.ts b/google-adk-agents/src/streaming/workflows.ts new file mode 100644 index 000000000..e76726063 --- /dev/null +++ b/google-adk-agents/src/streaming/workflows.ts @@ -0,0 +1,44 @@ +import type { LlmRequest } from '@google/adk'; +import { TemporalModel } from '@temporalio/google-adk-agents/workflow'; +import { condition, defineSignal, setHandler } from '@temporalio/workflow'; +import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; + +export const streamingTopic = 'responses'; + +export const consumerDoneSignal = defineSignal('consumer-done'); + +export async function streamingModelCall(prompt: string): Promise<{ text: string; chunks: number }> { + new WorkflowStream(); + + let consumerDone = false; + setHandler(consumerDoneSignal, () => { + consumerDone = true; + }); + + const model = new TemporalModel('gemini-2.5-flash', { + streamingTopic, + activity: { heartbeatTimeout: '5 seconds' }, + }); + + const request = { + model: 'gemini-2.5-flash', + contents: [{ role: 'user', parts: [{ text: prompt }] }], + config: {}, + toolsDict: {}, + liveConnectConfig: {}, + } as LlmRequest; + + let text = ''; + let chunks = 0; + // The turn's whole text is on the last, non-partial response; the deltas would double it. + for await (const response of model.generateContentAsync(request, true)) { + if (response.partial === true) { + chunks++; + continue; + } + text = (response.content?.parts ?? []).map((part) => part.text ?? '').join(''); + } + // Completing discards the stream log, racing a subscriber's final poll; the timeout covers no subscriber. + await condition(() => consumerDone, '10 seconds'); + return { text, chunks }; +} diff --git a/google-adk-agents/src/tools/README.md b/google-adk-agents/src/tools/README.md new file mode 100644 index 000000000..860bc5591 --- /dev/null +++ b/google-adk-agents/src/tools/README.md @@ -0,0 +1,23 @@ +# Google ADK Agents: Tools + +Exposes an existing Temporal Activity to the ADK agent as a tool with `activityAsTool`. When the model decides to call `getWeather`, the tool dispatches the registered `getWeather` Activity — durable and retriable — instead of running the I/O inside the Workflow body. + +## Run + +Run these from the `google-adk-agents/` root (run `npm install` there once first). + +```bash +# In one terminal, start the Worker (requires a local Temporal server and GEMINI_API_KEY): +GEMINI_API_KEY=... npx ts-node src/tools/worker.ts + +# In another terminal, run the scenario: +npx ts-node src/tools/client.ts +``` + +## Test + +```bash +npx mocha --exit --require ts-node/register --require source-map-support/register "src/tools/mocha/*.test.ts" +``` + +The tests run a real Worker against `TestWorkflowEnvironment`, driving `weatherAgent` end to end against a scripted model double. One takes the happy path: the model asks for `getWeather`, the Activity runs exactly once, and its result comes back on the next model turn. The other fails that Activity and asserts the failure reaches the model as the tool's response and the agent answers from it. No `GEMINI_API_KEY` is required. diff --git a/google-adk-agents/src/tools/activities.ts b/google-adk-agents/src/tools/activities.ts new file mode 100644 index 000000000..6b4c9fe02 --- /dev/null +++ b/google-adk-agents/src/tools/activities.ts @@ -0,0 +1,3 @@ +export async function getWeather(args: { city: string }): Promise { + return `The weather in ${args.city} is warm and sunny, 17 degrees.`; +} diff --git a/google-adk-agents/src/tools/client.ts b/google-adk-agents/src/tools/client.ts new file mode 100644 index 000000000..f277649ce --- /dev/null +++ b/google-adk-agents/src/tools/client.ts @@ -0,0 +1,21 @@ +import { Connection, Client } from '@temporalio/client'; +import { nanoid } from 'nanoid'; +import { weatherAgent } from './workflows'; + +async function run() { + const connection = await Connection.connect(); + const client = new Client({ connection }); + + const result = await client.workflow.execute(weatherAgent, { + taskQueue: 'google-adk-tools', + workflowId: 'google-adk-tools-' + nanoid(), + args: ['What is the weather in Tokyo?'], + }); + + console.log(result); +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/tools/mocha/workflows.test.ts b/google-adk-agents/src/tools/mocha/workflows.test.ts new file mode 100644 index 000000000..158bed6b9 --- /dev/null +++ b/google-adk-agents/src/tools/mocha/workflows.test.ts @@ -0,0 +1,97 @@ +import { BaseLlm } from '@google/adk'; +import type { BaseLlmConnection, LlmRequest, LlmResponse } from '@google/adk'; +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { ApplicationFailure } from '@temporalio/common'; +import { Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import { after, before, describe, it } from 'mocha'; +import assert from 'assert'; +import * as activities from '../activities'; +import { weatherAgent } from '../workflows'; + +// A fresh model instance per Activity invocation, so the turn has to come from the request. +function weatherModelProvider(): (model: string) => BaseLlm { + class WeatherLlm extends BaseLlm { + override async *generateContentAsync( + llmRequest: LlmRequest, + _stream = false, + _abortSignal?: AbortSignal, + ): AsyncGenerator { + const toolResponse = (llmRequest.contents ?? []) + .flatMap((content) => content.parts ?? []) + .find((part) => part.functionResponse?.name === 'getWeather')?.functionResponse?.response as + | { result?: unknown; error?: unknown } + | undefined; + if (toolResponse === undefined) { + yield { + content: { role: 'model', parts: [{ functionCall: { name: 'getWeather', args: { city: 'Tokyo' } } }] }, + turnComplete: true, + }; + return; + } + const text = + toolResponse.error === undefined + ? String(toolResponse.result) + : `I could not look up the weather: ${String(toolResponse.error)}`; + yield { content: { role: 'model', parts: [{ text }] }, turnComplete: true }; + } + + override async connect(_llmRequest: LlmRequest): Promise { + throw new Error('WeatherLlm does not support connect().'); + } + } + return (model: string) => new WeatherLlm({ model }); +} + +describe('google-adk-agents/tools workflow scenarios', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + async function runWeatherAgent(taskQueue: string, workflowId: string, workerActivities: typeof activities) { + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath: require.resolve('../workflows'), + activities: workerActivities, + plugins: [new GoogleAdkPlugin({ modelProvider: weatherModelProvider() })], + }); + return worker.runUntil( + testEnv.client.workflow.execute(weatherAgent, { + args: ['What is the weather in Tokyo?'], + workflowId, + taskQueue, + }), + ); + } + + it('weatherAgent: the model calls the tool, the Activity runs, and its result reaches the next turn', async () => { + const taskQueue = 'test-google-adk-tools-agent'; + const workflowId = taskQueue + '-' + Date.now(); + const result = await runWeatherAgent(taskQueue, workflowId, activities); + assert.strictEqual(result, 'The weather in Tokyo is warm and sunny, 17 degrees.'); + + const { events } = await testEnv.client.workflow.getHandle(workflowId).fetchHistory(); + const scheduled = (events ?? []).map((e) => e.activityTaskScheduledEventAttributes?.activityType?.name); + assert.strictEqual(scheduled.filter((name) => name === 'getWeather').length, 1); + assert.strictEqual(scheduled.filter((name) => name === 'adk-invokeModel').length, 2); + }); + + it('weatherAgent: a failed tool Activity reaches the model as the tool response, and the agent answers from it', async () => { + const taskQueue = 'test-google-adk-tools-agent-failure'; + const result = await runWeatherAgent(taskQueue, taskQueue + '-' + Date.now(), { + getWeather: async () => { + throw ApplicationFailure.nonRetryable('the weather service is down'); + }, + }); + assert.strictEqual(result, 'I could not look up the weather: Activity task failed'); + }); +}); diff --git a/google-adk-agents/src/tools/worker.ts b/google-adk-agents/src/tools/worker.ts new file mode 100644 index 000000000..b05c4fd26 --- /dev/null +++ b/google-adk-agents/src/tools/worker.ts @@ -0,0 +1,24 @@ +import { NativeConnection, Worker } from '@temporalio/worker'; +import { GoogleAdkPlugin } from '@temporalio/google-adk-agents'; +import * as activities from './activities'; + +async function run() { + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + try { + const worker = await Worker.create({ + connection, + taskQueue: 'google-adk-tools', + workflowsPath: require.resolve('./workflows'), + activities, + plugins: [new GoogleAdkPlugin()], + }); + await worker.run(); + } finally { + await connection.close(); + } +} + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/google-adk-agents/src/tools/workflows.ts b/google-adk-agents/src/tools/workflows.ts new file mode 100644 index 000000000..60f90df1a --- /dev/null +++ b/google-adk-agents/src/tools/workflows.ts @@ -0,0 +1,35 @@ +import { InMemoryRunner, LlmAgent, isFinalResponse, stringifyContent } from '@google/adk'; +import { Type } from '@google/genai'; +import { activityAsTool, TemporalModel } from '@temporalio/google-adk-agents/workflow'; + +export async function weatherAgent(prompt: string): Promise { + const agent = new LlmAgent({ + name: 'weather_agent', + model: new TemporalModel('gemini-2.5-flash'), + instruction: 'Use the getWeather tool to answer weather questions.', + tools: [ + activityAsTool({ + name: 'getWeather', + description: 'Get the current weather for a city.', + parameters: { + type: Type.OBJECT, + properties: { city: { type: Type.STRING, description: 'The city name' } }, + required: ['city'], + }, + }), + ], + }); + + const runner = new InMemoryRunner({ agent }); + + let finalText = ''; + for await (const event of runner.runEphemeral({ + userId: 'user', + newMessage: { role: 'user', parts: [{ text: prompt }] }, + })) { + if (isFinalResponse(event)) { + finalText = stringifyContent(event); + } + } + return finalText; +} diff --git a/google-adk-agents/tsconfig.json b/google-adk-agents/tsconfig.json new file mode 100644 index 000000000..488f2c62a --- /dev/null +++ b/google-adk-agents/tsconfig.json @@ -0,0 +1,13 @@ +{ + "extends": "@tsconfig/node22/tsconfig.json", + "version": "5.6.3", + "compilerOptions": { + "lib": ["es2021"], + "declaration": true, + "declarationMap": true, + "sourceMap": true, + "rootDir": "./src", + "outDir": "./lib" + }, + "include": ["src/**/*.ts"] +}