diff --git a/examples/java/adk/README.md b/examples/java/adk/README.md
new file mode 100644
index 000000000000..0596235d887f
--- /dev/null
+++ b/examples/java/adk/README.md
@@ -0,0 +1,109 @@
+# ADK AML Investigator Example
+
+This example demonstrates how to use the **Google Agent Development Kit (ADK)** with **Apache Beam** to build an AI-powered Anti-Money Laundering (AML) investigator.
+
+The pipeline monitors a **Cloud Spanner Change Stream** for new transactions and uses a **Gemini-powered Agent** to perform graph-based analysis (using Spanner Graph/GQL) to detect suspicious patterns.
+
+## Features
+
+- **Spanner Change Stream Integration**: Automatically triggers analysis on new `INSERT` events.
+- **ADK LlmAgent**: Uses Gemini 2.5 Flash to reason about transaction risks.
+- **Spanner Graph (GQL) Tools**: The agent is equipped with three GQL tools:
+    1. `detectCircularFlow`: Finds round-tripping loops (e.g., A -> B -> C -> A).
+    2. `detectFanInStructuring`: Finds smurfing patterns where multiple accounts funnel money to one collector.
+    3. `detectSharedIdentity`: Detects synthetic accounts sharing the same physical device.
+- **OpenTelemetry Tracing**: End-to-end observability of agent reasoning and tool execution.
+
+## Prerequisites
+
+- A Google Cloud Project.
+- `gcloud` CLI installed and authenticated.
+- Java 17 or higher.
+- Gradle.
+
+## Setup
+
+### 1. Spanner Database Setup
+
+Run the provided script to create a Spanner instance, database, schema, and seed data:
+
+```bash
+./examples/java/adk/setup-spanner.sh [INSTANCE_ID] [DATABASE_ID]
+```
+
+Defaults are `aml-instance` and `aml-db`.
+
+### 2. Run the Beam Pipeline
+
+#### Option A: Direct Runner (Local)
+Execute the pipeline locally:
+
+```bash
+JAVA_HOME=/Library/Java/JavaVirtualMachines/temurin-17.jdk/Contents/Home ./gradlew :examples:java:adk:execute --args=" \
+    --spannerInstance=aml-instance \
+    --spannerDatabase=aml-db \
+    --changeStreamName=TransactionsStream \
+    --outputTable=Transactions \
+    --project=radoslaws-playground-pso"
+```
+
+#### Option B: Dataflow Runner v1 (Legacy Worker)
+To run on Google Cloud Dataflow using the v1 runner (Legacy Worker), first build the worker JAR:
+
+```bash
+JAVA_HOME=/Library/Java/JavaVirtualMachines/temurin-17.jdk/Contents/Home ./gradlew :runners:google-cloud-dataflow-java:worker:shadowJar
+```
+
+Then, execute the pipeline:
+
+```bash
+JAVA_HOME=/Library/Java/JavaVirtualMachines/temurin-17.jdk/Contents/Home ./gradlew :examples:java:adk:execute --args=" \
+    --project=radoslaws-playground-pso \
+    --region=us-central1 \
+    --runner=DataflowRunner \
+    --spannerInstance=aml-instance \
+    --spannerDatabase=aml-db \
+    --changeStreamName=TransactionsStream \
+    --outputTable=Transactions \
+    --gcpTempLocation=gs://radoslaws-playground-pso/temp \
+    --tempLocation=gs://radoslaws-playground-pso/ \
+    --streaming=true \
+    --maxNumWorkers=1 \
+    --enableStreamingEngine \
+    --dataflowWorkerJar=runners/google-cloud-dataflow-java/worker/build/libs/beam-runners-google-cloud-dataflow-java-legacy-worker-2.76.0-SNAPSHOT.jar \
+    --experiments=disable_runner_v2,enable_otel_defaults \
+    --dataflowServiceOptions=enable_google_cloud_profiler,enable_google_cloud_heap_sampling"
+```
+
+## Demo Scenarios
+
+Once the pipeline is running, you can trigger AML scenarios by inserting "TRIGGER" transactions.
+
+### Scenario 1: Circular Flow (Round-Tripping)
+Trigger a loop where money goes Alice -> Bob -> Charlie -> Alice within 48 hours.
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_loop_03_TRIGGER', 'usr_charlie', 'usr_alice', 9500.00, 'PENDING', CURRENT_TIMESTAMP())"
+```
+
+### Scenario 2: Fan-In Structuring (Smurfing)
+Trigger a pattern where a third "mule" sends money to a collector who already received two sub-threshold transfers.
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_fan_03_TRIGGER', 'usr_bob', 'usr_collector', 9100.00, 'PENDING', CURRENT_TIMESTAMP())"
+```
+
+### Scenario 3: Shared Device (Co-location)
+Trigger a transfer between two accounts that are physically located on the same hardware device.
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_colocate_TRIGGER', 'usr_fraudA', 'usr_fraudB', 4500.00, 'PENDING', CURRENT_TIMESTAMP())"
+```
+
+## Observing Results
+
+Check the `Transactions` table in Spanner to see the `RiskReason` generated by the AI Agent:
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="SELECT TransactionId, RiskReason FROM Transactions WHERE TransactionId LIKE '%_TRIGGER'"
+```
diff --git a/examples/java/adk/build.gradle b/examples/java/adk/build.gradle
new file mode 100644
index 000000000000..263e2230d4b8
--- /dev/null
+++ b/examples/java/adk/build.gradle
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+    id 'java'
+    id 'org.apache.beam.module'
+}
+
+applyJavaNature(
+        exportJavadoc: false,
+        automaticModuleName: 'org.apache.beam.examples.adk',
+)
+provideIntegrationTestingDependencies()
+enableJavaPerformanceTesting()
+
+description = "Apache Beam :: Examples :: Java :: ADK"
+ext.summary = "Apache Beam SDK examples for ADK."
+
+dependencies {
+    implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
+    implementation project(":sdks:java:extensions:google-cloud-platform-core")
+    implementation project(path: ":sdks:java:core", configuration: "shadow")
+    implementation project(path: ":sdks:java:io:google-cloud-platform")
+    implementation project(path: ":runners:google-cloud-dataflow-java")
+    testImplementation library.java.google_cloud_pubsub
+    //runtimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+    runtimeOnly library.java.opentelemetry_exporter_otlp
+    runtimeOnly library.java.opentelemetry_extension_autoconfigure
+    runtimeOnly project(":sdks:java:extensions:opentelemetry-gcp-auth-extension")
+    // Google Cloud Spanner
+    implementation library.java.google_cloud_spanner
+
+    // OpenTelemetry
+    implementation library.java.opentelemetry_api
+    implementation library.java.opentelemetry_sdk
+
+    // ADK Dependencies
+    implementation "com.google.adk:google-adk:1.6.0"
+   // implementation "com.google.genai:google-genai:0.1.1"
+
+    implementation library.java.joda_time
+    implementation library.java.slf4j_api
+    implementation library.java.slf4j_simple
+}
+
+task execute (type:JavaExec) {
+    classpath = sourceSets.main.runtimeClasspath
+    mainClass = "org.apache.beam.examples.adk.AmlPipeline"
+}
diff --git a/examples/java/adk/revert-demo.sh b/examples/java/adk/revert-demo.sh
new file mode 100644
index 000000000000..b8a5113fb114
--- /dev/null
+++ b/examples/java/adk/revert-demo.sh
@@ -0,0 +1,17 @@
+#!/bin/bash
+
+# Configuration
+INSTANCE_ID=${1:-"aml-instance"}
+DATABASE_ID=${2:-"aml-db"}
+PROJECT_ID=$(gcloud config get-value project)
+
+echo "Using Project: $PROJECT_ID"
+echo "Using Instance: $INSTANCE_ID"
+echo "Using Database: $DATABASE_ID"
+
+echo "Reverting demo transactions..."
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="DELETE FROM Transactions WHERE TransactionId IN ('tx_loop_03_TRIGGER', 'tx_fan_03_TRIGGER', 'tx_colocate_TRIGGER');"
+
+echo "Demo reverted successfully. You can now re-trigger the scenarios."
diff --git a/examples/java/adk/setup-spanner.sh b/examples/java/adk/setup-spanner.sh
new file mode 100755
index 000000000000..c53baa09aba9
--- /dev/null
+++ b/examples/java/adk/setup-spanner.sh
@@ -0,0 +1,59 @@
+#!/bin/bash
+
+# Configuration
+INSTANCE_ID=${1:-"aml-instance"}
+DATABASE_ID=${2:-"aml-db"}
+PROJECT_ID=$(gcloud config get-value project)
+
+echo "Using Project: $PROJECT_ID"
+echo "Using Instance: $INSTANCE_ID"
+echo "Using Database: $DATABASE_ID"
+
+# 1. Create Instance (if not exists)
+if ! gcloud spanner instances describe "$INSTANCE_ID" > /dev/null 2>&1; then
+    echo "Creating Spanner instance $INSTANCE_ID..."
+    gcloud spanner instances create "$INSTANCE_ID" \
+        --config=regional-us-central1 \
+        --description="AML Demo Instance" \
+        --nodes=1 \
+        --edition=ENTERPRISE
+fi
+
+# 2. Create Database (if not exists)
+if ! gcloud spanner databases describe "$DATABASE_ID" --instance="$INSTANCE_ID" > /dev/null 2>&1; then
+    echo "Creating Spanner database $DATABASE_ID..."
+    gcloud spanner databases create "$DATABASE_ID" --instance="$INSTANCE_ID"
+fi
+
+# 3. Apply Schema
+echo "Applying DDL schema..."
+gcloud spanner databases ddl update "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --ddl-file="examples/java/adk/spanner-schema.sql"
+
+# 4. Seed Data
+echo "Seeding initial data..."
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO Account (AccountId, AccountHolder, Status, CreatedAt) VALUES ('usr_alice', 'Alice Smith', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_bob', 'Bob Jones', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_charlie', 'Charlie Brown', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_mule1', 'Mule One', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_mule2', 'Mule Two', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_collector', 'Collector Hub', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_fraudA', 'Fraud User A', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_fraudB', 'Fraud User B', 'CLEARED', CURRENT_TIMESTAMP());"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO SharedDevice (DeviceId, DeviceModel, FirstSeen) VALUES ('dev_hardware_xyz99', 'iPhone 15 Pro', CURRENT_TIMESTAMP());"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_loop_01', 'usr_alice', 'usr_bob', 10000.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 10 HOUR)), ('tx_loop_02', 'usr_bob', 'usr_charlie', 9800.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 HOUR));"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_fan_01', 'usr_mule1', 'usr_collector', 9200.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)), ('tx_fan_02', 'usr_mule2', 'usr_collector', 9400.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO AccountDevice (AccountId, DeviceId, LinkedAt) VALUES ('usr_fraudA', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)), ('usr_fraudB', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));"
+
+echo "Spanner setup complete."
+echo ""
+echo "To trigger Scenario 1 (Circular Flow), run:"
+echo "gcloud spanner databases execute-sql $DATABASE_ID --instance=$INSTANCE_ID --sql=\"INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_loop_03_TRIGGER', 'usr_charlie', 'usr_alice', 9500.00, 'PENDING', CURRENT_TIMESTAMP())\""
+echo ""
+echo "To trigger Scenario 2 (Fan-In), run:"
+echo "gcloud spanner databases execute-sql $DATABASE_ID --instance=$INSTANCE_ID --sql=\"INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_fan_03_TRIGGER', 'usr_bob', 'usr_collector', 9100.00, 'PENDING', CURRENT_TIMESTAMP())\""
+echo ""
+echo "To trigger Scenario 3 (Co-location), run:"
+echo "gcloud spanner databases execute-sql $DATABASE_ID --instance=$INSTANCE_ID --sql=\"INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_colocate_TRIGGER', 'usr_fraudA', 'usr_fraudB', 4500.00, 'PENDING', CURRENT_TIMESTAMP())\""
diff --git a/examples/java/adk/spanner-schema.sql b/examples/java/adk/spanner-schema.sql
new file mode 100644
index 000000000000..d5b3f081b1d5
--- /dev/null
+++ b/examples/java/adk/spanner-schema.sql
@@ -0,0 +1,68 @@
+-- =============================================================================
+-- 1. BASE RELATIONAL TABLES
+-- =============================================================================
+
+-- Accounts Node Table
+CREATE TABLE Account (
+    AccountId STRING(64) NOT NULL,
+    AccountHolder STRING(256) NOT NULL,
+    Status STRING(32) NOT NULL, -- 'PENDING', 'CLEARED', 'REVIEW_REQUIRED'
+    CreatedAt TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true)
+) PRIMARY KEY (AccountId);
+
+-- Shared Attributes Node Tables (For Synthetic/Co-location Checks)
+CREATE TABLE SharedDevice (
+    DeviceId STRING(128) NOT NULL,
+    DeviceModel STRING(128),
+    FirstSeen TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true)
+) PRIMARY KEY (DeviceId);
+
+-- Edge Table: Linking Accounts to Devices
+CREATE TABLE AccountDevice (
+    AccountId STRING(64) NOT NULL,
+    DeviceId STRING(128) NOT NULL,
+    LinkedAt TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true),
+    FOREIGN KEY (AccountId) REFERENCES Account (AccountId),
+    FOREIGN KEY (DeviceId) REFERENCES SharedDevice (DeviceId)
+) PRIMARY KEY (AccountId, DeviceId);
+
+-- Edge Table: Financial Transactions (Edges between Accounts)
+CREATE TABLE Transactions (
+    TransactionId STRING(64) NOT NULL,
+    SenderId STRING(64) NOT NULL,
+    ReceiverId STRING(64) NOT NULL,
+    Amount NUMERIC NOT NULL,
+    Status STRING(32) NOT NULL, -- 'PENDING', 'CLEARED', 'REVIEW_REQUIRED'
+    RiskReason STRING(MAX),
+    Timestamp TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true),
+    ReviewedAt TIMESTAMP,
+    FOREIGN KEY (SenderId) REFERENCES Account (AccountId),
+    FOREIGN KEY (ReceiverId) REFERENCES Account (AccountId)
+) PRIMARY KEY (TransactionId);
+
+-- Change Stream for Transactions
+CREATE CHANGE STREAM TransactionsStream FOR Transactions
+  OPTIONS (
+    exclude_update = true,
+    exclude_delete = true
+  );
+
+-- =============================================================================
+-- 2. SPANNER PROPERTY GRAPH DEFINITION
+-- =============================================================================
+
+CREATE PROPERTY GRAPH FinancialGraph
+  NODE TABLES (
+    Account,
+    SharedDevice
+  )
+  EDGE TABLES (
+    Transactions
+      SOURCE KEY (SenderId) REFERENCES Account (AccountId)
+      DESTINATION KEY (ReceiverId) REFERENCES Account (AccountId)
+      LABEL TRANSFERRED_TO,
+    AccountDevice
+      SOURCE KEY (AccountId) REFERENCES Account (AccountId)
+      DESTINATION KEY (DeviceId) REFERENCES SharedDevice (DeviceId)
+      LABEL USED_DEVICE
+  );
diff --git a/examples/java/adk/spanner-seed.sql b/examples/java/adk/spanner-seed.sql
new file mode 100644
index 000000000000..b9e8ceefa82c
--- /dev/null
+++ b/examples/java/adk/spanner-seed.sql
@@ -0,0 +1,29 @@
+-- Seed Accounts
+INSERT INTO Account (AccountId, AccountHolder, Status, CreatedAt) VALUES
+  ('usr_alice', 'Alice Smith', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_bob', 'Bob Jones', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_charlie', 'Charlie Brown', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_mule1', 'Mule One', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_mule2', 'Mule Two', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_collector', 'Collector Hub', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_fraudA', 'Fraud User A', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_fraudB', 'Fraud User B', 'CLEARED', CURRENT_TIMESTAMP());
+
+-- Seed Shared Device
+INSERT INTO SharedDevice (DeviceId, DeviceModel, FirstSeen) VALUES
+  ('dev_hardware_xyz99', 'iPhone 15 Pro', CURRENT_TIMESTAMP());
+
+-- Scenario 1: Circular Flow Setup
+INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES
+  ('tx_loop_01', 'usr_alice', 'usr_bob', 10000.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 10 HOUR)),
+  ('tx_loop_02', 'usr_bob', 'usr_charlie', 9800.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 HOUR));
+
+-- Scenario 2: Fan-In Structuring Setup
+INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES
+  ('tx_fan_01', 'usr_mule1', 'usr_collector', 9200.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)),
+  ('tx_fan_02', 'usr_mule2', 'usr_collector', 9400.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));
+
+-- Scenario 3: Shared Device Setup
+INSERT INTO AccountDevice (AccountId, DeviceId, LinkedAt) VALUES
+  ('usr_fraudA', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)),
+  ('usr_fraudB', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));
diff --git a/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlAgentDoFn.java b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlAgentDoFn.java
new file mode 100644
index 000000000000..208641d99e37
--- /dev/null
+++ b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlAgentDoFn.java
@@ -0,0 +1,267 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk;
+
+import com.google.adk.agents.LlmAgent;
+import com.google.adk.events.Event;
+import com.google.adk.models.Gemini;
+import com.google.adk.runner.InMemoryRunner;
+import com.google.adk.sessions.Session;
+import com.google.adk.tools.FunctionTool;
+import com.google.cloud.spanner.Mutation;
+import com.google.genai.Client;
+import com.google.genai.types.Content;
+import com.google.genai.types.FunctionResponse;
+import com.google.genai.types.Part;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.SdkHarnessOptions;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.values.KV;
+
+@SuppressWarnings("initialization.fields.uninitialized")
+public class AmlAgentDoFn extends DoFn<KV<String, TransactionEvent>, Mutation> {
+
+  private final String instanceId;
+  private final String databaseId;
+  private final String outputTable;
+
+  private transient SpannerGraphAmlTools graphTools;
+  private transient LlmAgent amlAgent;
+  private transient Tracer tracer;
+  private transient InMemoryRunner runner;
+  private transient Client client;
+
+  @StateId("memory")
+  private final StateSpec<ValueState<Integer>> valueStateSpec = StateSpecs.value(VarIntCoder.of());
+
+  public AmlAgentDoFn(String instanceId, String databaseId, String outputTable) {
+    this.instanceId = instanceId;
+    this.databaseId = databaseId;
+    this.outputTable = outputTable;
+  }
+
+  @Setup
+  public void setup(PipelineOptions options) {
+    // 1. Initialize DB and Spanner Graph Tools
+    this.graphTools = new SpannerGraphAmlTools();
+    this.graphTools.initSpanner(instanceId, databaseId);
+    this.tracer =
+        options
+            .as(SdkHarnessOptions.class)
+            .getOpenTelemetry()
+            .getTracer("org.apache.beam.examples.adk.aml");
+
+    // 2. Configure System Instructions for the Agent
+    String systemInstruction =
+        "You are an Anti-Money Laundering (AML) Investigator AI Agent.\n"
+            + "Given a transaction event, execute Spanner Graph (GQL) tools to uncover potential fraud rings:\n"
+            + "1. Run 'detectCircularFlow' using the sender ID to uncover round-tripping money loops.\n"
+            + "2. Run 'detectFanInStructuring' using the receiver ID to check for smurfing/fan-in aggregation.\n"
+            + "3. Run 'detectSharedIdentity' between sender and receiver to identify synthetic/co-located accounts.\n"
+            + "4. Consolidate findings and output a raw JSON object with exactly two string fields (no markdown formatting):\n"
+            + " - \"status\": The Risk Assessment ('CLEARED', 'REVIEW_REQUIRED', or 'PENDING').\n"
+            + " - \"riskReason\": The detailed justification.";
+
+    // 3. Build ADK LlmAgent with registered tools
+    FunctionTool circularFlowTool = FunctionTool.create(graphTools, "detectCircularFlow");
+    FunctionTool fanInTool = FunctionTool.create(graphTools, "detectFanInStructuring");
+    FunctionTool sharedIdentityTool = FunctionTool.create(graphTools, "detectSharedIdentity");
+    client =
+        Client.builder()
+            .vertexAI(true)
+            .project(options.as(GcpOptions.class).getProject())
+            .location("us")
+            .build();
+    this.amlAgent =
+        LlmAgent.builder()
+            .name("AmlGraphInvestigator")
+            .model(new Gemini("gemini-3.5-flash", client))
+            .description(
+                "Investigates financial transaction records using Spanner GQL Graph Queries.")
+            .instruction(systemInstruction)
+            .tools(circularFlowTool, fanInTool, sharedIdentityTool)
+            .build();
+
+    this.runner = new InMemoryRunner(this.amlAgent);
+  }
+
+  @Teardown
+  public void tearDown() {
+    if (graphTools != null) {
+      graphTools.close();
+    }
+    if (client != null) {
+      client.close();
+    }
+  }
+
+  private static void processAgentEvent(
+      Event event,
+      StringBuilder agentResponseBuilder,
+      AtomicBoolean toolCalledInTurn,
+      AtomicBoolean toolErroredInTurn) {
+    if (event.content().isPresent()) {
+      event
+          .content()
+          .get()
+          .parts()
+          .ifPresent(
+              parts -> {
+                for (Part part : parts) {
+                  if (part.text().isPresent()) {
+                    System.out.print(part.text().get());
+                    agentResponseBuilder.append(part.text().get());
+                  }
+                  if (part.functionCall().isPresent()) {
+                    toolCalledInTurn.set(true);
+                  }
+                  if (part.functionResponse().isPresent()) {
+                    FunctionResponse fr = part.functionResponse().get();
+                    fr.response()
+                        .ifPresent(
+                            responseMap -> {
+                              if (responseMap.containsKey("error")
+                                  || (responseMap.containsKey("status")
+                                      && "error"
+                                          .equalsIgnoreCase(
+                                              String.valueOf(responseMap.get("status"))))) {
+                                toolErroredInTurn.set(true);
+                              }
+                            });
+                  }
+                }
+              });
+    }
+    if (event.errorCode().isPresent() || event.errorMessage().isPresent()) {
+      toolErroredInTurn.set(true);
+    }
+  }
+
+  @ProcessElement
+  public void processElement(
+      @Element KV<String, TransactionEvent> pairOfUserAndTx,
+      @StateId("memory") ValueState<Integer> valueState,
+      MultiOutputReceiver out)
+      throws Exception {
+    TransactionEvent tx = pairOfUserAndTx.getValue();
+
+    Integer memory = valueState.read();
+    if (memory == null) {
+      memory = 0;
+    }
+    valueState.write(memory + 1);
+
+    Span parentSpan =
+        tracer
+            .spanBuilder("invoke_agent:AmlAgentWorker")
+            .setAttribute("transaction.id", tx.getTransactionId())
+            .startSpan();
+
+    try (Scope parentScope = parentSpan.makeCurrent()) {
+      String prompt =
+          String.format(
+              "Investigate TransactionID: %s, Sender: %s, Receiver: %s, Amount: $%.2f, Timestamp: %s",
+              tx.getTransactionId(),
+              tx.getSenderId(),
+              tx.getReceiverId(),
+              tx.getAmount(),
+              tx.getTimestamp());
+
+      // Create a unique session per transaction
+      Session session =
+          runner
+              .sessionService()
+              .createSession(amlAgent.name(), "user-" + tx.getSenderId(), null, null)
+              .blockingGet();
+
+      Content userMsg = Content.fromParts(Part.fromText(prompt));
+
+      final StringBuilder agentResponseBuilder = new StringBuilder();
+      final AtomicBoolean toolCalledInTurn = new AtomicBoolean(false);
+      final AtomicBoolean toolErroredInTurn = new AtomicBoolean(false);
+
+      // ADK's LlmAgent executes reasoning, tool execution loops, and final output automatically
+      runner
+          .runAsync("user-" + tx.getSenderId(), session.id(), userMsg)
+          .blockingForEach(
+              event ->
+                  processAgentEvent(
+                      event, agentResponseBuilder, toolCalledInTurn, toolErroredInTurn));
+
+      String finalAssessment = agentResponseBuilder.toString();
+
+      String parsedStatus = "REVIEW_REQUIRED";
+      String parsedRiskReason = finalAssessment;
+
+      try {
+        com.fasterxml.jackson.databind.ObjectMapper mapper =
+            new com.fasterxml.jackson.databind.ObjectMapper();
+        // Remove potential markdown formatting returned by LLMs
+        String cleanedJson = finalAssessment.replaceAll("```json", "").replaceAll("```", "").trim();
+        com.fasterxml.jackson.databind.JsonNode jsonNode = mapper.readTree(cleanedJson);
+        if (jsonNode.has("status")) {
+          parsedStatus = jsonNode.get("status").asText();
+        }
+        if (jsonNode.has("riskReason")) {
+          parsedRiskReason = jsonNode.get("riskReason").asText();
+        }
+      } catch (Exception ex) {
+        System.err.println("Failed to parse agent response as JSON: " + finalAssessment);
+      }
+
+      // Update transaction with RiskReason and ReviewedAt
+      Mutation mutation =
+          Mutation.newUpdateBuilder(outputTable)
+              .set("TransactionId")
+              .to(tx.getTransactionId())
+              .set("SenderId")
+              .to(tx.getSenderId())
+              .set("ReceiverId")
+              .to(tx.getReceiverId())
+              .set("Amount")
+              .to(tx.getAmount())
+              .set("Status")
+              .to(parsedStatus)
+              .set("Timestamp")
+              .to(com.google.cloud.Timestamp.parseTimestamp(tx.getTimestamp()))
+              .set("RiskReason")
+              .to(parsedRiskReason)
+              .set("ReviewedAt")
+              .to(com.google.cloud.Timestamp.now())
+              .build();
+      System.out.println("mutating: " + mutation);
+      out.get(AmlPipeline.MUTATION_TAG).output(mutation);
+      out.get(AmlPipeline.REVIEW_TAG).output(mutation.toString());
+
+    } catch (Exception e) {
+      parentSpan.recordException(e);
+      throw e;
+    } finally {
+      parentSpan.end();
+    }
+  }
+}
diff --git a/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlChangeStreamFilterDoFn.java b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlChangeStreamFilterDoFn.java
new file mode 100644
index 000000000000..21c203cc1dd9
--- /dev/null
+++ b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlChangeStreamFilterDoFn.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.math.BigDecimal;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.DataChangeRecord;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.ModType;
+import org.apache.beam.sdk.transforms.DoFn;
+
+public class AmlChangeStreamFilterDoFn extends DoFn<DataChangeRecord, TransactionEvent> {
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  @ProcessElement
+  public void processElement(@Element DataChangeRecord record, OutputReceiver<TransactionEvent> out)
+      throws Exception {
+    if (record.getModType() == ModType.INSERT) {
+      for (Mod mod : record.getMods()) {
+        String keysJson = mod.getKeysJson();
+        String newValuesJson = mod.getNewValuesJson();
+        if (keysJson != null && newValuesJson != null) {
+          JsonNode keys = MAPPER.readTree(keysJson);
+          JsonNode newValues = MAPPER.readTree(newValuesJson);
+          System.out.println("values " + newValuesJson);
+          String transactionId =
+              keys.has("TransactionId") ? keys.get("TransactionId").asText() : "";
+          String senderId = newValues.has("SenderId") ? newValues.get("SenderId").asText() : "";
+          String receiverId =
+              newValues.has("ReceiverId") ? newValues.get("ReceiverId").asText() : "";
+          BigDecimal amount =
+              newValues.has("Amount")
+                  ? new BigDecimal(newValues.get("Amount").asText())
+                  : BigDecimal.ZERO;
+          String status = newValues.has("Status") ? newValues.get("Status").asText() : null;
+          String riskReason =
+              newValues.has("RiskReason") ? newValues.get("RiskReason").asText() : null;
+          String timestamp = newValues.has("Timestamp") ? newValues.get("Timestamp").asText() : "";
+          String reviewedAt =
+              newValues.has("ReviewedAt") ? newValues.get("ReviewedAt").asText() : null;
+          TransactionEvent value =
+              TransactionEvent.create(
+                  transactionId,
+                  senderId,
+                  receiverId,
+                  amount,
+                  status,
+                  riskReason,
+                  timestamp,
+                  reviewedAt);
+          System.out.println("new transaction - " + value);
+
+          out.output(value);
+        }
+      }
+    }
+  }
+}
diff --git a/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlPipeline.java b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlPipeline.java
new file mode 100644
index 000000000000..455d93da76c3
--- /dev/null
+++ b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlPipeline.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk;
+
+import com.google.cloud.spanner.Mutation;
+import com.google.cloud.spanner.SpannerOptions;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.SerializableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+
+public class AmlPipeline {
+  public static final TupleTag<Mutation> MUTATION_TAG = new TupleTag<Mutation>() {};
+  public static final TupleTag<String> REVIEW_TAG = new TupleTag<String>() {};
+
+  public static void main(String[] args) {
+    AmlPipelineOptions options =
+        PipelineOptionsFactory.fromArgs(args).withValidation().as(AmlPipelineOptions.class);
+
+    Pipeline pipeline = Pipeline.create(options);
+    SpannerOptions.enableOpenTelemetryTraces();
+    SpannerOptions.disableOpenCensusMetrics();
+    SpannerOptions.enableOpenTelemetryMetrics();
+    SpannerConfig spannerConfig =
+        SpannerConfig.create()
+            .withProjectId(options.getProject())
+            .withInstanceId(options.getSpannerInstance())
+            .withDatabaseId(options.getSpannerDatabase());
+    pipeline
+        .apply(
+            "Add transaction",
+            PubsubIO.readMessages()
+                .withEnableOpenTelemetryTracing()
+                .fromSubscription("projects/radoslaws-playground-pso/subscriptions/txn-sub"))
+        .apply("Redistribute", Redistribute.arbitrarily())
+        .apply("Create mutation", ParDo.of(new NewTransactionDoFn(options.getOutputTable())))
+        .apply(
+            "Write New Tx to Spanner Table",
+            SpannerIO.write()
+                .withSpannerConfig(spannerConfig)
+                .withEnableOpenTelemetryTracing(true));
+    PCollectionTuple amlResult =
+        pipeline
+            .apply(
+                "Read Spanner Change Stream",
+                SpannerIO.readChangeStream()
+                    .withSpannerConfig(spannerConfig)
+                    .withLowLatency()
+                    .withEnableOpenTelemetryTracing(true)
+                    .withChangeStreamName(options.getChangeStreamName())
+                    .withInclusiveStartAt(
+                        com.google.cloud.Timestamp.ofTimeSecondsAndNanos(
+                            com.google.cloud.Timestamp.now().getSeconds(), 0)))
+            .apply("Filter New Transactions", ParDo.of(new AmlChangeStreamFilterDoFn()))
+            //   .apply("Redistribute", Redistribute.arbitrarily())
+            .apply("Key", WithKeys.<String, TransactionEvent>of(TransactionEvent::getSenderId))
+            .setCoder(
+                KvCoder.of(StringUtf8Coder.of(), SerializableCoder.of(TransactionEvent.class)))
+            .apply(
+                "ADK Graph AML Analysis",
+                ParDo.of(
+                        new AmlAgentDoFn(
+                            options.getSpannerInstance(),
+                            options.getSpannerDatabase(),
+                            options.getOutputTable()))
+                    .withOutputTags(MUTATION_TAG, TupleTagList.of(REVIEW_TAG)));
+    amlResult
+        .get(REVIEW_TAG)
+        .apply(
+            "Write to pubsub",
+            PubsubIO.writeStrings()
+                .withEnableOpenTelemetryTracing()
+                .to("projects/radoslaws-playground-pso/topics/review"));
+    amlResult
+        .get(MUTATION_TAG)
+        .apply(
+            "Write to Spanner Table",
+            SpannerIO.write()
+                .withSpannerConfig(spannerConfig)
+                .withEnableOpenTelemetryTracing(true));
+
+    pipeline.run().waitUntilFinish();
+  }
+}
diff --git a/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlPipelineOptions.java b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlPipelineOptions.java
new file mode 100644
index 000000000000..fb53d7e87803
--- /dev/null
+++ b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/AmlPipelineOptions.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk;
+
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.Validation.Required;
+
+public interface AmlPipelineOptions extends GcpOptions, PipelineOptions {
+
+  @Description("Cloud Spanner Instance ID")
+  @Required
+  String getSpannerInstance();
+
+  void setSpannerInstance(String value);
+
+  @Description("Cloud Spanner Database ID")
+  @Required
+  String getSpannerDatabase();
+
+  void setSpannerDatabase(String value);
+
+  @Description("Cloud Spanner Change Stream Name")
+  @Required
+  String getChangeStreamName();
+
+  void setChangeStreamName(String value);
+
+  @Description("Target Spanner Table for AML Risk Output")
+  @Required
+  String getOutputTable();
+
+  void setOutputTable(String value);
+}
diff --git a/examples/java/adk/src/main/java/org/apache/beam/examples/adk/NewTransactionDoFn.java b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/NewTransactionDoFn.java
new file mode 100644
index 000000000000..b4a67108f04d
--- /dev/null
+++ b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/NewTransactionDoFn.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.cloud.spanner.Mutation;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import java.math.BigDecimal;
+import java.util.Objects;
+import java.util.UUID;
+import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.SdkHarnessOptions;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.util.Preconditions;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NewTransactionDoFn extends DoFn<PubsubMessage, Mutation> {
+
+  private final String outputTable;
+  @MonotonicNonNull ObjectMapper mapper = null;
+  private transient @MonotonicNonNull Tracer tracer = null;
+  private static final Logger LOG = LoggerFactory.getLogger(NewTransactionDoFn.class);
+
+  public NewTransactionDoFn(String outputTable) {
+    this.outputTable = outputTable;
+  }
+
+  // POJO representing the incoming JSON structure
+  public static class TransactionPayload {
+    @JsonProperty("TransactionId")
+    private @Nullable String transactionId;
+
+    @JsonProperty("SenderId")
+    private @Nullable String senderId;
+
+    @JsonProperty("ReceiverId")
+    private @Nullable String receiverId;
+
+    @JsonProperty("Amount")
+    private double amount;
+
+    @JsonProperty("Timestamp")
+    private @Nullable String timestamp;
+
+    // Getters and Setters
+    public @Nullable String getTransactionId() {
+      return transactionId;
+    }
+
+    public void setTransactionId(String transactionId) {
+      this.transactionId = transactionId;
+    }
+
+    public @Nullable String getSenderId() {
+      return senderId;
+    }
+
+    public void setSenderId(String senderId) {
+      this.senderId = senderId;
+    }
+
+    public @Nullable String getReceiverId() {
+      return receiverId;
+    }
+
+    public void setReceiverId(String receiverId) {
+      this.receiverId = receiverId;
+    }
+
+    public double getAmount() {
+      return amount;
+    }
+
+    public void setAmount(double amount) {
+      this.amount = amount;
+    }
+
+    public @Nullable String getTimestamp() {
+      return timestamp;
+    }
+
+    public void setTimestamp(String timestamp) {
+      this.timestamp = timestamp;
+    }
+  }
+
+  @Setup
+  public void setup(PipelineOptions options) {
+    mapper = new ObjectMapper();
+    this.tracer =
+        options
+            .as(SdkHarnessOptions.class)
+            .getOpenTelemetry()
+            .getTracer("org.apache.beam.examples.adk.aml");
+  }
+
+  @ProcessElement
+  public void processElement(@Element PubsubMessage in, OutputReceiver<Mutation> out)
+      throws Exception {
+    Span parentSpan =
+        Preconditions.checkStateNotNull(tracer)
+            .spanBuilder("NewTransaction.Process")
+            .setAttribute("msgId", in.getMessageId())
+            .startSpan();
+
+    try (Scope ignored = parentSpan.makeCurrent()) {
+      String payload = new String(in.getPayload(), java.nio.charset.StandardCharsets.UTF_8);
+
+      // Deserialize JSON payload directly into the TransactionPayload POJO
+      TransactionPayload tx =
+          Preconditions.checkStateNotNull(mapper).readValue(payload, TransactionPayload.class);
+      // Update transaction with RiskReason and ReviewedAt
+      Mutation mutation =
+          Mutation.newInsertOrUpdateBuilder(outputTable)
+              .set("TransactionId")
+              .to(UUID.randomUUID().toString())
+              .set("SenderId")
+              .to(tx.getSenderId())
+              .set("ReceiverId")
+              .to(tx.getReceiverId())
+              .set("Amount")
+              .to(new BigDecimal(tx.getAmount()))
+              .set("Status")
+              .to("PENDING")
+              .set("Timestamp")
+              .to(
+                  com.google.cloud.Timestamp.parseTimestamp(
+                      Objects.requireNonNullElse(tx.getTimestamp(), "")))
+              .build();
+      LOG.info("mutating {}", mutation);
+      out.output(mutation);
+
+    } catch (Exception e) {
+      parentSpan.recordException(e);
+      throw e;
+    } finally {
+      parentSpan.end();
+    }
+  }
+}
diff --git a/examples/java/adk/src/main/java/org/apache/beam/examples/adk/SpannerGraphAmlTools.java b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/SpannerGraphAmlTools.java
new file mode 100644
index 000000000000..361a1c5c2123
--- /dev/null
+++ b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/SpannerGraphAmlTools.java
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk;
+
+import com.google.adk.tools.Annotations;
+import com.google.cloud.spanner.*;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+@SuppressWarnings("initialization.field.uninitialized")
+public class SpannerGraphAmlTools implements Serializable {
+
+  private transient DatabaseClient dbClient;
+  private transient Spanner spanner;
+
+  public void initSpanner(String instanceId, String databaseId) {
+
+    SpannerOptions options =
+        SpannerOptions.newBuilder()
+            .setOpenTelemetry(GlobalOpenTelemetry.get())
+            .setEnableEndToEndTracing(true)
+            .setEnableExtendedTracing(true)
+            .build();
+
+    this.spanner = options.getService();
+    DatabaseId db = DatabaseId.of(options.getProjectId(), instanceId, databaseId);
+    this.dbClient = spanner.getDatabaseClient(db);
+  }
+
+  public void close() {
+    if (spanner != null) {
+      spanner.close();
+    }
+  }
+
+  // --- Tool 1: Circular Flow ---
+  public String detectCircularFlow(@Annotations.Schema(name = "userId") String userId) {
+    String gqlQuery =
+        "GRAPH FinancialGraph "
+            + "MATCH (a:Account {AccountId: @user_id})-[t1:TRANSFERRED_TO]->(b:Account) "
+            + "      -[t2:TRANSFERRED_TO]->(c:Account) "
+            + "      -[t3:TRANSFERRED_TO]->(a) "
+            + "WHERE TIMESTAMP_DIFF(t3.Timestamp, t1.Timestamp, HOUR) <= 48 "
+            + "RETURN b.AccountId AS Hop1, c.AccountId AS Hop2, t1.Amount AS InitialAmount, t3.Amount AS ReturnedAmount";
+
+    Statement statement = Statement.newBuilder(gqlQuery).bind("user_id").to(userId).build();
+
+    List<String> findings = new ArrayList<>();
+    try (ResultSet rs = dbClient.singleUse().executeQuery(statement)) {
+      while (rs.next()) {
+        findings.add(
+            String.format(
+                "Loop detected via %s -> %s (Initial: $%.2f, Returned: $%.2f)",
+                rs.getString("Hop1"),
+                rs.getString("Hop2"),
+                rs.getBigDecimal("InitialAmount"),
+                rs.getBigDecimal("ReturnedAmount")));
+      }
+    }
+
+    return findings.isEmpty()
+        ? "NO_CIRCULAR_FLOW"
+        : "SUSPICIOUS_CIRCULAR_FLOW: " + String.join("; ", findings);
+  }
+
+  // --- Tool 2: Fan-In Structuring ---
+  public String detectFanInStructuring(
+      @Annotations.Schema(name = "targetUserId") String targetUserId) {
+    String gqlQuery =
+        "select * from GRAPH_TABLE(\n"
+            + "FinancialGraph \n"
+            + "MATCH (sender:Account)-[t:TRANSFERRED_TO]->(collector:Account {AccountId: @target_user_id}) \n"
+            + "      WHERE t.Timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY) \n"
+            + "            AND t.Amount BETWEEN 2000 AND 9999 \n"
+            + "      WITH collector, \n"
+            + "           COUNT(DISTINCT sender) AS unique_senders, \n"
+            + "           SUM(t.Amount) AS total_funneled \n"
+            + "      \n"
+            + "      RETURN unique_senders, total_funneled ) as gt where unique_senders >= 3 ";
+
+    Statement statement =
+        Statement.newBuilder(gqlQuery).bind("target_user_id").to(targetUserId).build();
+
+    try (ResultSet rs = dbClient.singleUse().executeQuery(statement)) {
+      if (rs.next()) {
+        return String.format(
+            "SUSPICIOUS_FAN_IN: Funneled by %d unique accounts, Total: $%.2f",
+            rs.getLong("unique_senders"), rs.getBigDecimal("total_funneled"));
+      }
+    }
+
+    return "NO_FAN_IN_DETECTED";
+  }
+
+  // --- Tool 3: Shared Identity ---
+  public String detectSharedIdentity(
+      @Annotations.Schema(name = "senderId") String senderId,
+      @Annotations.Schema(name = "receiverId") String receiverId) {
+    String gqlQuery =
+        "GRAPH FinancialGraph "
+            + "MATCH (acc1:Account {AccountId: @sender_id})-[t:TRANSFERRED_TO]->(acc2:Account {AccountId: @receiver_id}) "
+            + "MATCH (acc1)-[:USED_DEVICE]->(shared_node)<-[:USED_DEVICE]-(acc2) "
+            + "RETURN LABELS(shared_node)[0] AS SharedAttributeType, shared_node.DeviceId AS SharedAttributeValue";
+
+    Statement statement =
+        Statement.newBuilder(gqlQuery)
+            .bind("sender_id")
+            .to(senderId)
+            .bind("receiver_id")
+            .to(receiverId)
+            .build();
+
+    List<String> sharedElements = new ArrayList<>();
+    try (ResultSet rs = dbClient.singleUse().executeQuery(statement)) {
+      while (rs.next()) {
+        sharedElements.add(
+            rs.getString("SharedAttributeType") + ":" + rs.getString("SharedAttributeValue"));
+      }
+    }
+
+    return sharedElements.isEmpty()
+        ? "NO_SHARED_IDENTITY"
+        : "SYNTHETIC_CLUSTER_ALERT: Shared " + String.join(", ", sharedElements);
+  }
+}
diff --git a/examples/java/adk/src/main/java/org/apache/beam/examples/adk/TransactionEvent.java b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/TransactionEvent.java
new file mode 100644
index 000000000000..509483927f5b
--- /dev/null
+++ b/examples/java/adk/src/main/java/org/apache/beam/examples/adk/TransactionEvent.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk;
+
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import java.math.BigDecimal;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+
+@AutoValue
+@DefaultSchema(AutoValueSchema.class)
+public abstract class TransactionEvent implements Serializable {
+  public abstract String getTransactionId();
+
+  public abstract String getSenderId();
+
+  public abstract String getReceiverId();
+
+  public abstract BigDecimal getAmount();
+
+  @javax.annotation.Nullable
+  public abstract String getStatus();
+
+  @javax.annotation.Nullable
+  public abstract String getRiskReason();
+
+  public abstract String getTimestamp();
+
+  @javax.annotation.Nullable
+  public abstract String getReviewedAt();
+
+  public static TransactionEvent create(
+      String transactionId,
+      String senderId,
+      String receiverId,
+      BigDecimal amount,
+      @javax.annotation.Nullable String status,
+      @javax.annotation.Nullable String riskReason,
+      String timestamp,
+      @javax.annotation.Nullable String reviewedAt) {
+    return new AutoValue_TransactionEvent(
+        transactionId, senderId, receiverId, amount, status, riskReason, timestamp, reviewedAt);
+  }
+}
diff --git a/examples/java/adk/src/test/java/org/apache/beam/examples/adk/SimpleJsonPublisher.java b/examples/java/adk/src/test/java/org/apache/beam/examples/adk/SimpleJsonPublisher.java
new file mode 100644
index 000000000000..aced04d908da
--- /dev/null
+++ b/examples/java/adk/src/test/java/org/apache/beam/examples/adk/SimpleJsonPublisher.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.examples.adk; /*
+                                       * Licensed to the Apache Software Foundation (ASF) under one
+                                       * or more contributor license agreements.  See the NOTICE file
+                                       * distributed with this work for additional information
+                                       * regarding copyright ownership.  The ASF licenses this file
+                                       * to you under the Apache License, Version 2.0 (the
+                                       * "License"); you may not use this file except in compliance
+                                       * with the License.  You may obtain a copy of the License at
+                                       *
+                                       *     http://www.apache.org/licenses/LICENSE-2.0
+                                       *
+                                       * Unless required by applicable law or agreed to in writing, software
+                                       * distributed under the License is distributed on an "AS IS" BASIS,
+                                       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+                                       * See the License for the specific language governing permissions and
+                                       * limitations under the License.
+                                       */
+
+import com.google.cloud.pubsub.v1.Publisher;
+import com.google.protobuf.ByteString;
+import com.google.pubsub.v1.PubsubMessage;
+import com.google.pubsub.v1.TopicName;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+public class SimpleJsonPublisher {
+
+  public static void main(String[] args) throws Exception {
+    // Replace with your GCP Project ID and Pub/Sub Topic ID
+    String projectId = "radoslaws-playground-pso";
+    String topicId = "txn";
+
+    TopicName topicName = TopicName.of(projectId, topicId);
+    Publisher publisher = null;
+
+    try {
+
+      System.setProperty("google.cloud.project", "radoslaws-playground-pso");
+      System.setProperty("otel.exporter.otlp.endpoint", "https://telemetry.googleapis.com");
+      System.setProperty("otel.traces.exporter", "otlp");
+      System.setProperty("otel.java.global-autoconfigure.enabled", "true");
+      System.setProperty("otel.traces.sampler.arg", "1.00");
+      System.setProperty("otel.service.name", "TRANSACTION_PRODUCER");
+      // Creates a Cloud Trace exporter.
+
+      OpenTelemetry ignored = GlobalOpenTelemetry.get();
+      // Create a publisher instance bound to the topic
+      publisher =
+          Publisher.newBuilder(topicName)
+              .setOpenTelemetry(ignored)
+              .setEnableOpenTelemetryTracing(true)
+              .build();
+
+      // Sample JSON payload matching the Spanner AML transaction format
+      String jsonPayload =
+          "{\n"
+              + "  \"TransactionId\": \""
+              + UUID.randomUUID().toString()
+              + "\",\n"
+              + "  \"SenderId\": \"usr_charlie\",\n"
+              + "  \"ReceiverId\": \"usr_alice\",\n"
+              + "  \"Amount\": 9500.00,\n"
+              + "  \"Timestamp\": \"2026-07-31T12:00:00Z\"\n"
+              + "}";
+
+      // Convert JSON string to Pub/Sub ByteString message
+      ByteString data = ByteString.copyFrom(jsonPayload, StandardCharsets.UTF_8);
+      PubsubMessage pubsubMessage =
+          PubsubMessage.newBuilder()
+              .setData(data)
+              .putAttributes("contentType", "application/json") // Optional metadata
+              .build();
+
+      // Publish the message and wait for the message ID callback
+      String messageId = publisher.publish(pubsubMessage).get();
+      System.out.println("Published JSON message successfully with ID: " + messageId);
+
+    } finally {
+      if (publisher != null) {
+        // Shut down the publisher and release resources
+        publisher.shutdown();
+        publisher.awaitTermination(1, TimeUnit.MINUTES);
+      }
+    }
+    Thread.sleep(10000);
+  }
+}
diff --git a/settings.gradle.kts b/settings.gradle.kts
index a7cdfc705152..ab8ec4799a70 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -86,6 +86,7 @@ include(":release:go-licenses:py")
 
 include(":examples:java")
 include(":examples:java:twitter")
+include(":examples:java:adk")
 include(":examples:java:cdap")
 include(":examples:java:cdap:hubspot")
 include(":examples:java:cdap:salesforce")