agent-engine-sdk-langgraph

A LangChain SDK for MongoDB Atlas Agent Engine. Provides a thin LangChain-specific wrapper over the agent-engine-runner-shared platform runtime.

Quick Start

Installation

pip install agent-engine-sdk-langgraph

Or in a uv project:

uv add agent-engine-sdk-langgraph

Minimal Agent

from agent_engine_sdk_langgraph import App
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

app = App(app_name="my-agent", app_version="1.0.0")

@app.tool()
def lookup(query: str) -> str:
    """Search the knowledge base."""
    return "result for " + query

@app.entrypoint
def build_agent():
    from langchain_openai import ChatOpenAI

    llm = app.llm(ChatOpenAI(model="gpt-5.4"))
    tools = app.get_tools()

    def call_model(state: MessagesState):
        response = llm.invoke(state["messages"])
        return {"messages": [response]}

    graph = StateGraph(MessagesState)
    graph.add_node("agent", call_model)
    graph.add_node("tools", ToolNode(tools))
    graph.set_entry_point("agent")
    graph.add_edge("tools", "agent")
    return graph.compile(checkpointer=app.checkpointer())

app.run()

Using Memory

app.memory is the unified agent-engine-sdk-memory Memory facade (app-bound over the platform runtime). Identity resolves per call from the argument, a bound context, or the ambient execution context.

app = App(app_name="my-agent")

# Save a semantic fact — returns CreateSemanticResult
result = app.memory.save_semantic(
    text="User prefers dark mode",
    label="pref-theme",
    user_id="u1",
    metadata={"channel": "web", "priority": "high"},  # optional caller-supplied metadata
)
if result.acknowledged:
    ...

# Search — returns list[MemoryChunk]
chunks = app.memory.search_semantic(query="user preferences", user_id="u1")
for chunk in chunks:
    print(chunk.content)

# Build prompt context — returns ContextResponse
# Default sources are LTM (episodic + semantic); pass
# enabled_sources={"stm", ...} to include recent turns.
context = app.memory.build_context(query="help me", user_id="u1")
prompt_block = context.formatted_context  # str (or structured list, depending on format)

build_context_from_sources (per-source retrieval modes, filters, and top_k) is available on the ambient app.memory runtime. It returns the full ContextResponse, so the per-source metadata (ranking_strategy, source_outcomes) survives; the platform stamps tenancy and carries the response back through the durable execution. session_id is required only when the stm source is requested.

from agent_engine_sdk_langgraph import Memory re-exports the same class as agent_engine_sdk_memory.Memory. Constructing Memory(api_key=...) / Memory(base_url=...) yourself is the HTTP/direct path — not ambient app-bound.

When this breaks

This is a hard cut: there is no dual API and no compatibility shim on app.memory. Customer agents break only when they rebuild an agent image that picks up runner-base wheels containing this SDK. Platform merge alone, or redeploying an old agent image, does not change the SDK code already baked into that image. There is no stored-memory data migration — only client return shapes and call conventions change.

Migrating from the pre-facade surface

• Return types / truthiness / metadata / build_context: writes return typed results (CreateSemanticResult, CreateEpisodicResult, etc.) with acknowledged — not bare bool/dicts. Prefer if result.acknowledged: (not if result: — Pydantic models are always truthy). Similarity searches return list[MemoryChunk] (chunk.content, optional chunk.similarity_score). Loose fields that used to be top-level dict keys (title, summary, tags, term, definition, related_terms, …) live under chunk.metadata (e.g. ep.metadata.get("title") instead of ep.get("title")). build_context returns ContextResponse; use context.formatted_context, not the return value as a string.
• Write helpers are keyword-only (save_semantic(text=..., label=..., …)). Positional calls raise TypeError.
• Reads with visibility="private" keep the ambient user filter (previously passing any visibility argument dropped it): private-visibility searches now return only the current user's memories unless an explicit user_id is passed.
• top_k: per-source search_semantic / search_episodes / search_taxonomic default to top_k=50 (was often 10). discover_procedures still defaults to 10. Unified Memory.search() still defaults to top_k=10. Public build_context has no top_k parameter. Use max_tokens for a gross context-construction budget (not a fetch cost). After retrieval and ranking, the server subtracts a 500-token formatting reserve, then greedily selects whole memory chunks that fit in the remainder. Positive values at or below 500 leave no budget for memories. Values above 500 can still yield empty context when no chunk fits. Pass an explicit top_k on search helpers if you need the old limit.
• Identity: required fields that cannot be resolved raise MemoryIdentityError (no more soft None / silent skip). save_episode requires a resolvable session_id (ambient invocation context is fine; otherwise pass it explicitly or bind a MemoryRequestContext). Blank values do not count as set.
• App-bound create fidelity: create-result id may be "" and has_embedding is typically False — gate success on .acknowledged, not id. Details per operation are in the memory package capability matrix.
• Gets vs search: get_semantic / get_taxonomic_term / list_episodes still return loosely-typed dicts on app-bound. Only search* methods return list[MemoryChunk].
• Renames: if anyone called the pre-facade names, use the facade public API — create_taxonomic → save_taxonomic; list_taxonomic_domains → list_domains.

Before / after

# save_semantic: bool → .acknowledged; positional → keyword-only
# before
ok = app.memory.save_semantic("User prefers dark mode", "pref-theme", user_id="u1")
if ok:
    ...
# after
result = app.memory.save_semantic(
    text="User prefers dark mode",
    label="pref-theme",
    user_id="u1",
)
if result.acknowledged:
    ...

# save_episode: str|None → CreateEpisodicResult (.acknowledged / .id)
# before
doc_id = app.memory.save_episode(title="Quote chat", content=summary, user_id="u1")
if doc_id:
    ...
# after
episode = app.memory.save_episode(
    title="Quote chat",
    content=summary,
    user_id="u1",
    metadata={"channel": "web", "priority": "high"},  # optional caller-supplied metadata
    # session_id from ambient context, or pass explicitly
)
if episode.acknowledged:
    print(episode.id)  # may be "" on app-bound

# search_episodes: dict.get → MemoryChunk metadata + content; pin top_k if needed
# before
episodes = app.memory.search_episodes(query="policy quote", top_k=10)
for ep in episodes:
    print(ep.get("title"), ep.get("content"))
# after
episodes = app.memory.search_episodes(query="policy quote", top_k=10)
for ep in episodes:
    print(ep.metadata.get("title"), ep.content)

# build_context: str → ContextResponse.formatted_context
# before
prompt = app.memory.build_context(query="help me", user_id="u1")
# after
context = app.memory.build_context(query="help me", user_id="u1")
prompt = context.formatted_context

Reference migration (in the Agent Engine examples repository):

• agents/insurance-agent/src/insurance_agent/main.py

Per-backend capability differences and app-bound gaps are documented in the memory package capability matrix. Method-level docs for the Memory facade live in agent-engine-sdk-memory.

Enable memory with features.memory: true in agent.yaml, or with the legacy ENABLE_MEMORY=true environment variable when that feature flag is omitted. Existing apps may still pass enable_memory=... or enable_tracing=... to App(...), but those constructor flags are deprecated: move memory into agent.yaml, and remove enable_tracing entirely because tracing is always on.

Skills

Skills are named markdown files (SKILL.md) that the LLM can load on demand via progressive disclosure. Use them to encode domain expertise (e.g. security review rules, coding conventions) that would bloat the system prompt if always included.

Directory layout

my_agent/
  skills/
    security-checklist/
      SKILL.md
    style-guide/
      SKILL.md

Pass the parent skills/ directory to skills=[...]. At runtime, deepagents lists that directory through the configured backend and discovers each immediate child directory containing SKILL.md as one skill. Discovery is one level deep, not recursive.

SKILL.md frontmatter

---
name: security-checklist
description: Security review rules for Python code, focusing on injection and auth
---

# Security Checklist

## REVIEW-RULE-ID-SEC-1: SQL injection
Never concatenate user input into SQL...

## REVIEW-RULE-ID-SEC-2: Command / path injection
Calls to `subprocess.run`, `os.system`, `shell=True`, and `open()` must not
interpolate untrusted input...

deepagents validates skill frontmatter at runtime. It skips unreadable or unparsable frontmatter and skills missing name or description; Agent Skills naming or directory-name violations produce warnings but may still load. The SDK forwards declared paths without inspecting or filtering them.

Wiring skills into your agent

Prerequisite: agent.yaml must include features.deep_agent: true or App.deep_agent() will raise RuntimeError at construction time:

features:
  deep_agent: true

Dependency: deep-agent graphs need the optional runtime extra: pip install "agent-engine-sdk-langgraph[deepagents]" (or uv add "agent-engine-sdk-langgraph[deepagents]"). The extra pins the tested deepagents version so applications track SDK releases instead of the upstream library. Applications that never call App.deep_agent() don't install it.

from langchain_openai import ChatOpenAI
from agent_engine_sdk_langgraph import App

app = App(app_name="My Reviewer")

@app.entrypoint
def build_agent():
    return app.deep_agent(
        llm=ChatOpenAI(model="gpt-5.4"),
        system_prompt="You are a code reviewer.",
        skills=["skills"],
    )

app.run()

Each skills=[...] entry is a parent source directory, not a leaf skill directory or a SKILL.md file. Paths are relative to the directory containing agent.yaml, so skills=["skills"] works for both single-agent images (/app/skills) and monorepo images (/app/<agent-subdirectory>/skills). If your skills live elsewhere inside the agent source tree, set AGENTIC_SKILLS_DIR to that relative directory and make skills=[...] relative to it. At runtime, deepagents reads frontmatter from each discovered skill and passes its metadata (name, description, and resolved path) to the LLM as a skills system block in the system prompt.

Progressive disclosure

On turn 1, the LLM sees only skill metadata — not the bodies. When a user asks about security, the LLM decides to read_file("<agent-dir>/skills/security-checklist/SKILL.md"), and the full body arrives as a ToolMessage in context for turn 2.

This keeps the base prompt lean (metadata is roughly 50 tokens per skill) while allowing deep expertise to be loaded on demand.

Bundled skill files and sandboxing

The ToolPod's writable filesystem and shell handlers still use WORKSPACE_DIR, which defaults to /tmp/agent-workspace. Keep that as scratch space.

Bundled skills are treated as read-only resources instead. The ToolPod derives the default skills root from AGENTIC_AGENT_CONFIG_PATH or AGENTIC_AGENT_WORKDIR: if the runtime config is /app/agent.yaml, the skills root is /app/skills; if the runtime config is /app/agents/reviewer/agent.yaml, the skills root is /app/agents/reviewer/skills. AGENTIC_SKILLS_DIR overrides that root and must be relative to the agent source root. Read-only filesystem tools can load files under that root without setting WORKSPACE_DIR to the skills directory. Write, edit, and shell operations still stay in the writable workspace. The skills root is resolved at Tool Pod startup, not at SDK import time, so a normal static SDK import works — no import-order workaround is needed.

Subagent non-inheritance

Skills are visible only to the agent that declares them. If your agent spawns subagents (via the task tool), those subagents do not inherit the parent's skills. Pass skills=[...] on each subagent spec that needs skill files.

Reserved tool names

The deep-agent runtime reserves 9 tool names for built-ins. Do not register @app.tool() with any of these names — it silently shadows the built-in and breaks skills/sandbox behavior:

• read_file, write_file, edit_file, ls, glob, grep (filesystem)
• execute (shell)
• write_todos (planning)
• task (subagent dispatch)

Picking a name that collides will silently shadow the built-in — there's no import-time error.

Size guidance

Target < 200 lines per SKILL.md body. Larger skills:

• Consume more context when loaded (each read_file is a full-body dump)
• Risk hitting the LLM's single-message context limit on complex turns
• Suggest the skill should be split into multiple focused files

Tip: reset threads after editing SKILL.md

The runtime caches skills_metadata in agent state for the lifetime of a thread. If you edit a SKILL.md file, existing threads will continue using the stale metadata until reset. In dev: delete the thread or start a fresh session. In production: skill changes should be paired with a new model/prompt version rollout.

Full minimal example

See the Code Reviewer Agent in Agent Engine examples repository for a reference implementation:

• Agent Engine examples repository agents/code-reviewer-agent/src/code_reviewer_agent/main.py — wiring
• Agent Engine examples repository agents/code-reviewer-agent/skills/*/SKILL.md — example skills

Streaming

LangGraphBaseAgent.stream() yields StreamEvent objects. Iterate with async for to receive token-level updates, subagent lifecycle markers, and the final result.

event	When it fires	data fields
token	Each LLM-token chunk from the root agent or any active subagent.	content: token text. source: "" for the root agent, or the subagent's graph name. tool_call_id: parent's task tool_call_id when one is in flight for this subagent (may be "" until assembled).
subagent_start	A subagent run begins. Emitted from one of two paths: (1) primary — the parent agent's task tool_call is observed; (2) synthetic fallback — a sourced token arrives before the parent tool_call has assembled (buffered-dispatch providers). The two paths dedup against each other so exactly one start fires per subagent per turn.	source / subagent_name: the subagent's graph name. tool_call_id: the parent's task tool_call_id, or "" if the synthetic fallback fired before the tool_call assembled. description: the description arg the parent passed to task (empty when synthetic fires first).
subagent_end	A subagent run finishes. Primary path: the parent graph observes a Command-close whose tool_call_id matches an open subagent. Defensive path: stream errors or completes with dangling subagents — one subagent_end is emitted per orphan so consumers can close their UI state.	source / subagent_name: same as the start. tool_call_id: the closing tool_call_id, or "" for orphan ends from a synthetic-only start that never received a real tool_call. summary: the subagent's final ToolMessage content (empty for defensive ends).
result	Final completion of the root agent.	response plus the full message list.
suspend	HITL interrupt — the graph is paused awaiting human review.	suspend_payload, checkpoint_id.

Notes for consumers:

• subagent_start and subagent_end are always paired, including on abnormal termination. The cleanup arm in stream() distinguishes GeneratorExit (consumer disconnect — drops state without yielding, since no consumer remains) from provider-side errors (yields defensive subagent_end then re-raises).
• For parallel dispatches of the same subagent type, each invocation has its own tool_call_id. Prefer tool_call_id for routing tokens and fall back to source only when tool_call_id == "".
• subagent_end.summary is the subagent's final response text — the same string the parent agent will see as the task tool's return value.

Durable native interrupts

Durable workflows replay LangGraph's native interrupt() call and translate its newly generated native id to the previously recorded OE activity position. See Durable LangGraph interrupt and resume for the complete suspension, replay, and Command(resume=...) flow and its code call sites.

API Reference

See docs/api.md for the auto-generated App / LangGraph surface. For the Memory facade (methods, return types, identity rules), see agent-engine-sdk-memory and its capability matrix.

Configuration

Environment variable	Default	Description
MDB_AGENTIC_STORE_DB	mdb_store	Base name for the per-project MongoDB store used for LangGraph checkpoints (AER mode only). Project scoping/discovery still applies unless overridden below.
CHECKPOINT_DB_NAME	(unset)	Exact MongoDBSaver database name when set. Skips project scoping and discovery. Opt-in for dual-runtime shared checkpoint DBs (set on the agent AER pod env / SecretRefs).
CHECKPOINTER_SERVER_SELECTION_TIMEOUT	5.0	Seconds for MongoDB checkpointer server selection before checkpoint IO fails.
CHECKPOINTER_CONNECT_TIMEOUT	5.0	Seconds for MongoDB checkpointer connection establishment.
CHECKPOINTER_SOCKET_TIMEOUT	15.0	Seconds for MongoDB checkpointer socket reads/writes.

By default the LangGraph checkpoint thread_id is session_id:workspace_id. Agents can override this with @app.resolve_thread_id (return value used verbatim on fresh and resume). Custom keys are invisible to Atlas Agent Engine /query/sessions* history lookups, which still use only the default session/workspace-derived keys. Agents that bypass workspace scoping also own collision isolation within the checkpoint database.

LangGraph time-travel would patch the source thread_id. Atlas Agent Engine instead creates a new session. Native-checkpoint sessions copy a completed checkpoint onto a new thread; durable-workflow sessions branch from OE-validated state reconstructed in fenced scratch. See Session fork vs LangGraph time-travel.

The shared Tool and LLM replay model is described in Durable activity identity. See Durable compiled subgraphs and Durable Deep Agent delegation for complete sequence diagrams, examples, and code call-site maps. Only effects routed through Atlas Agent Engine's secure LLM and Tool wrappers participate in durable record/replay. For durable Tool replay, the ToolCall must come from the secure LLM wrapper and execute through a Tool returned by app.get_tools().

Durable workflows do not expose LangGraph's dynamic Send fan-out. The normal platform checkpointer rejects application-authored Send writes. Graphs created by app.deep_agent() are an opaque exception because LangChain uses Send internally to route ToolCalls; this private compatibility does not make Send a supported application API. Use fixed graph edges, compiled subgraphs, or Deep Agent task delegation instead. Native-checkpoint workflows are unaffected.

Reads are scoped-only. Session history expands each Atlas Agent Engine session_id to only its workspace-scoped composite key; the bare unscoped key is never queried once a workspace scope is known, because bare keys are readable and writable by every workspace on the shared store. Legacy checkpoints written before scoping existed are therefore no longer served by the history endpoints; do not re-add the fallback. An empty scope is legitimate only on explicitly unscoped runtimes (local dev / tests, no APP_ID). Managed AERs carry REQUIRE_PROJECT_SCOPED_DB; if APP_ID is missing there, both reads and writes fail closed instead of trusting the wire workspace or using bare keys. Production adopters of custom keys should still treat checkpoint-key uniqueness inside a shared DB as agent-owned.

Development

Requirements

• Python >= 3.11
• uv

Dev Setup

uv sync --extra dev

Testing

For the same checks CI runs (lint + format + pyright + tests), use the unified runner: ./scripts/test.sh agent-engine-sdk-langgraph from the repo root.

uv run pytest

Type Checking

uv run pyright

Regenerate API Docs

make docs

Linting & Formatting

# Check for lint errors
uv run ruff check src

# Auto-fix lint errors
uv run ruff check --fix src

# Format code
uv run ruff format src