Google ADK + Neo4j Integration

pip install "google-adk>=2.0.0" neo4j-mcp-server "neo4j-agent-memory[google-adk,vertex-ai]" spacy
python -m spacy download en_core_web_sm

pip install "google-adk>=2.0.0" neo4j-graphrag sentence-transformers

NEO4J_URI=neo4j+s://your-instance.databases.neo4j.io
NEO4J_DATABASE=neo4j
NEO4J_TRANSPORT_MODE=http
NEO4J_MCP_HTTP_HOST=127.0.0.1
NEO4J_MCP_HTTP_PORT=8443
NEO4J_READ_ONLY=true          # write-cypher is not advertised to clients

# Memory Graph
MEMORY_NEO4J_URI=neo4j+s://your-memory-instance.databases.neo4j.io
MEMORY_NEO4J_USERNAME=neo4j
MEMORY_NEO4J_PASSWORD=your-password
MEMORY_NEO4J_DATABASE=neo4j

# Google Cloud Settings (Required for Vector Embeddings)
GCP_PROJECT_ID=your-gcp-project-id
GCP_LOCATION=us-central1

gcloud auth application-default login
gcloud auth application-default set-quota-project YOUR_PROJECT_ID
gcloud services enable aiplatform.googleapis.com --project YOUR_PROJECT_ID

import base64
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

# Create HTTP Basic Auth header
credentials = base64.b64encode(b"username:password").decode()

# Bind the running MCP Server to Google ADK
mcp_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="http://localhost:8443/mcp",
        headers={"Authorization": f"Basic {credentials}"}
    ),
    # Allow-list the tools the agent may see. Pairs with NEO4J_READ_ONLY
    # on the server, and shortens the tool schema sent with every prompt.
    tool_filter=["get-schema", "read-cypher"],
)

from google.adk import Agent
from google.adk.apps import App
from google.adk.plugins import ReflectAndRetryToolPlugin
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from neo4j_agent_memory import MemoryClient, MemorySettings
from neo4j_agent_memory.config.settings import Neo4jConfig, ExtractionConfig, ExtractorType

# 1. Configure the Memory Graph Connection
settings = MemorySettings(
    neo4j=Neo4jConfig(
        uri="neo4j+s://memory-instance...",
        username="neo4j",
        password="password"
    ),
    extraction=ExtractionConfig(extractor_type=ExtractorType.SPACY)
)
memory_client = MemoryClient(settings)
await memory_client.connect()

# 2. Construct the Agent
agent = Agent(
    model="gemini-3-flash",
    name="neo4j_explorer",
    instruction="You are a helpful graph database assistant.",
    tools=[mcp_tools, PreloadMemoryTool()]  # Combine MCP capabilities with Memory recall
)

# 3. Wrap it in an App. ReflectAndRetryToolPlugin feeds tool errors back to the
#    model, which lets it correct Cypher that references a wrong label or property.
app = App(
    name="neo4j_app",
    root_agent=agent,
    plugins=[ReflectAndRetryToolPlugin(max_retries=3)],
)

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from neo4j_agent_memory.integrations.google_adk import Neo4jMemoryService

session_service = InMemorySessionService()
session = await session_service.create_session(state={}, app_name=app.name, user_id="user_1")

# Initialize the ADK-compliant Memory Service
neo4j_memory_service = Neo4jMemoryService(
    memory_client=memory_client,
    user_id="user_1",
)

# Attach the memory service to the Runner. In ADK 2.0 the Runner takes the App container.
runner = Runner(
    app=app,
    session_service=session_service,
    memory_service=neo4j_memory_service
)

# Run the prompt
content = types.Content(role='user', parts=[types.Part(text="Who did Google invest in?")])
events = runner.run_async(session_id=session.id, user_id="user_1", new_message=content)

async for event in events:
    if hasattr(event, 'content') and event.content:
        for part in event.content.parts:
            if part.text:
                print(part.text)

# Save the conversational state and extracted entities back to Neo4j
fresh_session = await session_service.get_session(session_id=session.id, app_name=app.name, user_id="user_1")
await neo4j_memory_service.add_session_to_memory(fresh_session)

from google.adk import Agent, Event, Workflow

def validate(node_input: str | None = None):
    cypher = (node_input or "").strip()
    try:
        driver.execute_query(f"EXPLAIN {cypher}", database_="neo4j")
    except Exception:
        return Event(route=["INVALID"])
    return Event(route=["VALID"])

workflow = Workflow(
    name="text2cypher",
    edges=[
        ("START", cypher_writer_agent, validate),
        (validate, {"VALID": run_query, "INVALID": reject}),
    ],
)

from google.adk.tools.function_tool import FunctionTool
from neo4j_graphrag.embeddings import SentenceTransformerEmbeddings
from neo4j_graphrag.retrievers import HybridCypherRetriever

# `node` and `score` are in scope: continue into the graph from each vector hit.
RETRIEVAL_QUERY = """
WITH node AS chunk, score
MATCH (article:Article)-[:HAS_CHUNK]->(chunk)
OPTIONAL MATCH (article)-[:MENTIONS]->(org:Organization)
RETURN chunk.text AS text, article.id AS article_id, article.title AS title,
       collect(DISTINCT org.name)[..5] AS companies, score
"""

retriever = HybridCypherRetriever(
    driver=driver,
    vector_index_name="news_sbert",
    fulltext_index_name="news_fulltext",
    retrieval_query=RETRIEVAL_QUERY,
    embedder=SentenceTransformerEmbeddings(),
)

def search_news(question: str) -> str:
    """Search news by meaning and return the source article and companies mentioned.

    Args:
        question: A natural-language description of what to look for.
    """
    result = retriever.search(query_text=question, top_k=5)
    return json.dumps([item.content for item in result.items])

agent = Agent(model="gemini-3-flash", name="news_analyst",
              tools=[FunctionTool(search_news)])