Use with the OpenAI Agents SDK

Run an actual agents.Agent through Runner.run, then save and verify its conversation and task trace using the Neo4j adapter. This page chooses the Agents SDK orchestration API; Chat Completions function-schema dictionaries are a separate interface.

1. Prepare the selected environment

Use Python 3.10+ and a POSIX shell. The recipe uses BoltSettings to select Neo4j explicitly and disables entity extraction so that message persistence can be checked independently. Use a dedicated AuraDB instance with vector-index support and no incompatible existing vectors. Follow the Aura connection setup and copy its connection values into the exports below. These recipes use the published Python 0.7.0 package; provider access and database execution must be verified in your environment.

Create a local folder and virtual environment for the examples on this page. The commands reuse an existing environment without changing its files:

mkdir -p ~/agent-memory-tutorials
cd ~/agent-memory-tutorials
if [ -e .venv ]; then
  printf '%s\n' 'Using the existing virtual environment.'
else
  python3 -m venv .venv
fi
source .venv/bin/activate

Expected: ~/agent-memory-tutorials is your working directory and its virtual environment is active. Install the published SDK with the command below.

Each complete code block labelled Save as names a file to create in this folder using your editor. Copy the entire block, including imports and the entry point. Expand each helper disclosure and use Copy code to copy its full source. Keep all files together so their imports resolve.

When continuing from another tutorial or guide, retain the existing environment, configuration, session files, and .tutorial-state/. Reuse unchanged helper files; compare an existing file before replacing it, and finish any pending cleanup or recovery before changing the code that owns its state.

python -m pip install 'neo4j-agent-memory[openai-agents]==0.7.0' openai-agents
export NEO4J_URI='neo4j+s://<instance-id>.databases.neo4j.io'
export NEO4J_USERNAME='neo4j'
export NEO4J_PASSWORD='replace-with-your-Aura-password'
export NEO4J_DATABASE='neo4j'
export OPENAI_API_KEY='replace-with-your-provider-key'
export OPENAI_MODEL='replace-with-an-available-model-id'

The OpenAI-backed recipes require access to text-embedding-3-small; it produces 1536-dimensional vectors in this configuration. Configure the selected chat model separately where required. Credentials are read from the environment, never printed by the programs.

2. Read the complete recipe

Create the subdirectory below, then save both complete files under the displayed names. Run commands from agent-memory-tutorials/; Python resolves common from the script’s integrations/ directory.

mkdir -p integrations

The script imports this shared helper from the same directory. It supplies explicit database settings, closes clients through the calling context manager, and fails when expected records are absent. Task traces record the observable outcome of the call; they do not expose hidden model reasoning.

Complete integrations/common.py
Save as integrations/common.py
"""Shared Aura connection through Bolt and readback for integration recipes."""

import os


def settings(embedding=None):
    from neo4j_agent_memory import BoltSettings
    from neo4j_agent_memory.llm import from_provider

    return BoltSettings(
        neo4j={
            "uri": os.environ["NEO4J_URI"],
            "username": os.environ["NEO4J_USERNAME"],
            "password": os.environ["NEO4J_PASSWORD"],
            "database": os.getenv("NEO4J_DATABASE", "neo4j"),
        },
        embedding=embedding or from_provider("openai/text-embedding-3-small", kind="embedding"),
        extraction={"extractor_type": "none"},
    )


async def verify_messages(client, session_id, expected):
    conversation = await client.short_term.get_conversation(session_id)
    contents = [message.content for message in conversation.messages]
    if not all(text in contents for text in expected):
        raise RuntimeError(f"Message readback failed for {session_id}")
    print(f"Verified stored messages; session={session_id}")


async def recorded_turn(client, session_id, prompt, respond):
    """Record a task outcome; this does not claim to capture hidden model reasoning."""
    trace = await client.reasoning.start_trace(session_id=session_id, task=prompt)
    try:
        await client.short_term.add_message(session_id, "user", prompt, extract_entities=False)
        context = await client.get_context(prompt, session_id=session_id)
        reply = str(await respond(context))
        if not reply.strip():
            raise RuntimeError("The framework returned no text")
        await client.short_term.add_message(session_id, "assistant", reply, extract_entities=False)
        await verify_messages(client, session_id, [prompt, reply])
    except Exception as exc:
        await client.reasoning.complete_trace(
            trace.id, success=False, outcome=f"Run failed: {type(exc).__name__}"
        )
        raise
    await client.reasoning.complete_trace(
        trace.id, success=True, outcome="Reply stored and read back"
    )
    stored = await client.reasoning.get_trace_with_steps(trace.id)
    if stored is None or stored.success is not True:
        raise RuntimeError(f"Trace readback failed for {trace.id}")
    print(f"Verified task trace: {trace.id}")
    return reply
Complete integrations/openai_agents_recipe.py
Save as integrations/openai_agents_recipe.py
"""Assemble an actual Agents SDK run with explicit message persistence."""

import asyncio
import os
from uuid import uuid4

from common import settings, verify_messages


# tag::function_tool[]
def make_recall_tool(memory):
    from agents import function_tool

    @function_tool
    async def recall_context(query: str) -> str:
        """Retrieve prior conversation context relevant to `query`."""
        return await memory.get_context(query)

    return recall_context


# end::function_tool[]


async def exercise(client, run):
    from neo4j_agent_memory.integrations.openai_agents import Neo4jOpenAIMemory, record_agent_trace

    session_id = f"docs-openai-{uuid4().hex[:8]}"
    memory = Neo4jOpenAIMemory(memory_client=client, session_id=session_id)
    if hasattr(run, "bind_memory"):
        run.bind_memory(memory)
    prompt = "Suggest a simple project planning checklist."
    await memory.save_message("user", prompt, extract_entities=False)
    messages = [{"role": "user", "content": prompt}]
    try:
        context = await memory.get_context(prompt)
        reply = str(await run(prompt, context))
        if not reply.strip():
            raise RuntimeError("The Agents SDK returned no text")
        await memory.save_message("assistant", reply, extract_entities=False)
        messages.append({"role": "assistant", "content": reply})
        await verify_messages(client, session_id, [prompt, reply])
    except Exception as exc:
        await record_agent_trace(
            memory,
            messages=messages,
            task=prompt,
            outcome=f"Run failed: {type(exc).__name__}",
            success=False,
        )
        raise
    trace = await record_agent_trace(
        memory, messages=messages, task=prompt, outcome="Reply stored and read back", success=True
    )
    stored = await client.reasoning.get_trace_with_steps(trace.id)
    if stored is None or stored.success is not True:
        raise RuntimeError(f"Trace readback failed for {trace.id}")
    print(f"Verified task trace: {trace.id}")
    return reply


async def main():
    from agents import Agent, Runner

    from neo4j_agent_memory import MemoryClient

    memory_holder = []

    async def run(prompt, context):
        agent = Agent(
            name="Checklist assistant",
            model=os.environ["OPENAI_MODEL"],
            instructions=f"Answer briefly. Retrieved background:\n{context}",
            tools=[make_recall_tool(memory_holder[0])],
        )
        return (await Runner.run(agent, prompt)).final_output

    run.bind_memory = memory_holder.append

    async with MemoryClient(settings()) as client:
        print(await exercise(client, run))


if __name__ == "__main__":
    asyncio.run(main())

3. Run and verify persistence

python integrations/openai_agents_recipe.py

Expected: stored-message and task-trace verification followed by a nonempty final_output. The user turn is saved before the agent runs. A provider failure records a failed outcome and propagates; a successful trace is written only after persistence checks.

Troubleshooting and cleanup

Stop when the program raises an exception; a printed model response alone does not verify persistence. Check the configured database, provider access and vector dimensions before retrying. The program prints the session or trace IDs needed for inspection and closes the client. Retained example records remain in the dedicated database; follow the Aura tutorial’s cleanup procedure for that dedicated instance only when you no longer need them. Reusing a session groups records; it does not authorize a user to read them.

Adapt the integration

The repository’s [openai-agents] extra installs the underlying openai client used by its adapter. The command above also installs the separate openai-agents distribution that supplies agents.Agent and Runner.

Neo4jOpenAIMemory provides get_context, save_message and get_conversation. It is not an automatic Agents SDK session implementation: this recipe explicitly orchestrates those operations. record_agent_trace consumes normalized message dictionaries and actual tool-call records when supplied; it does not directly consume every SDK event type.

create_memory_tools(memory) returns Chat Completions-style function-schema dictionaries. Do not pass those dictionaries directly as Agents SDK function tools. To expose memory operations to an Agents SDK agent, write @function_tool wrappers around the async adapter methods with explicit arguments and return types, then add those functions to Agent(tools=…​). Only register write tools the application intends the agent to use. integrations/openai_agents_recipe.py ships this factory alongside the recipe, wrapping the adapter’s async get_context in exactly that pattern:

def make_recall_tool(memory):
    from agents import function_tool

    @function_tool
    async def recall_context(query: str) -> str:
        """Retrieve prior conversation context relevant to `query`."""
        return await memory.get_context(query)

    return recall_context

main() in integrations/openai_agents_recipe.py registers the wrapper with Agent(tools=[make_recall_tool(…​)]). The model decides whether to call a registered tool, so a successful run does not show that recall_context was called; the recipe’s own checks cover the explicit get and save calls.

For Chat Completions, dispatch returned tool calls with execute_memory_tool(memory, tool_name, arguments) and retain matched call IDs and responses in conversation history. Import it from the adapter’s memory module; the package root does not export it. This fragment assumes a connected memory and a tool_call taken from a Chat Completions response:

import json

from neo4j_agent_memory.integrations.openai_agents.memory import execute_memory_tool

result = await execute_memory_tool(
    memory,
    tool_call.function.name,
    json.loads(tool_call.function.arguments),
)

execute_memory_tool is a coroutine function. Await it; it returns a JSON string to send back as the tool message. Merely sending a schema to a model does not execute the function. Provider selection is covered by Configure providers. Product search and payment operations remain application code.

Adapter reference

This section lists the adapter surface of the published 0.7.0 package. Neo4jOpenAIMemory, create_memory_tools, record_agent_trace and llm_provider_from_openai_agents are exported from neo4j_agent_memory.integrations.openai_agents.

Neo4jOpenAIMemory(memory_client, session_id, user_id=None) provides these async methods:

Method Returns

get_context(query, max_items=10, …​)

A context string; include_short_term, include_long_term and include_reasoning select the layers.

save_message(role, content, …​)

The stored message. Optional tool_calls, tool_call_id, extract_entities and generate_embedding.

get_conversation(limit=50, include_system=True)

The session’s messages as role/content dictionaries.

search(query, limit=10, …​)

Matching messages, entities and preferences as dictionaries.

add_preference(category, preference)

The stored preference.

search_preferences(query, category=None, limit=10)

Matching preferences as dictionaries.

clear_session()

Nothing; removes the session’s messages.

Chat Completions tools

create_memory_tools(memory) returns four function schemas. execute_memory_tool(memory, tool_name, arguments) runs the named tool with the decoded argument dictionary.

Tool Arguments Purpose

search_memory

query, limit

Search messages, entities and preferences.

save_preference

category, preference

Store a user preference. This is a write tool.

recall_preferences

query, category

Retrieve stored preferences.

search_entities

query, entity_type, limit

Search the entity graph; entity_type is one of PERSON, LOCATION, ORGANIZATION, EVENT or OBJECT.

Traces

record_agent_trace(memory, messages, task, tool_calls=None, outcome=None, success=True) stores a task trace, as in the recipe. To reuse past outcomes, import the lookup helpers from the adapter’s tracing module:

from neo4j_agent_memory.integrations.openai_agents.tracing import (
    format_traces_for_prompt,
    get_similar_traces,
)

traces = await get_similar_traces(memory, task="Plan a project kickoff", limit=3)
background = format_traces_for_prompt(traces)

format_traces_for_prompt returns an empty string when no traces match. llm_provider_from_openai_agents(model) passes a configured model to memory extraction.

See also

See Backend capabilities and scoping before adding hosted or multi-user behavior. Neo4j Agent Memory is a Neo4j Labs project with community support.