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") def lookup(query: str) -> str: """Search the knowledge base.""" return "result for " + query 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 <../agent-engine-sdk-memory/README.md>`__ 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.) withacknowledged— not barebool/dicts. Preferif result.acknowledged:(notif result:— Pydantic models are always truthy). Similarity searches returnlist[MemoryChunk](chunk.content, optionalchunk.similarity_score). Loose fields that used to be top-level dict keys (title,summary,tags,term,definition,related_terms, …) live underchunk.metadata(e.g.ep.metadata.get("title")instead ofep.get("title")).build_contextreturnsContextResponse; usecontext.formatted_context, not the return value as a string.Write helpers are keyword-only (
save_semantic(text=..., label=..., …)). Positional calls raiseTypeError.Reads with ``visibility="private"`` keep the ambient user filter (previously passing any
visibilityargument dropped it): private-visibility searches now return only the current user’s memories unless an explicituser_idis passed.``top_k``: per-source
search_semantic/search_episodes/search_taxonomicdefault totop_k=50(was often10).discover_proceduresstill defaults to10. UnifiedMemory.search()still defaults totop_k=10. Publicbuild_contexthas notop_kparameter. Usemax_tokensfor 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 explicittop_kon search helpers if you need the old limit.Identity: required fields that cannot be resolved raise
MemoryIdentityError(no more softNone/ silent skip).save_episoderequires a resolvablesession_id(ambient invocation context is fine; otherwise pass it explicitly or bind aMemoryRequestContext). Blank values do not count as set.App-bound create fidelity: create-result
idmay be""andhas_embeddingis typicallyFalse— gate success on.acknowledged, notid. Details per operation are in the memory package capability matrix.Gets vs search:
get_semantic/get_taxonomic_term/list_episodesstill return loosely-typed dicts on app-bound. Onlysearch*methods returnlist[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
Note
Prerequisite: agent.yaml must include features.deep_agent: true or App.deep_agent() will raise RuntimeError at construction time:
features: deep_agent: true
from langchain_openai import ChatOpenAI from agent_engine_sdk_langgraph import App app = App(app_name="My Reviewer") 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
tasktool), those subagents do not inherit the parent’s skills. Passskills=[...]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_fileis 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— wiringAgent 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 |
|---|---|---|
| Each LLM-token chunk from the root agent or any active subagent. |
|
| A subagent run begins. Emitted from one of two paths: (1) primary — the parent agent’s |
|
| A subagent run finishes. Primary path: the parent graph observes a |
|
| Final completion of the root agent. |
|
| HITL interrupt — the graph is paused awaiting human review. |
|
Notes for consumers:
subagent_startandsubagent_endare always paired, including on abnormal termination. The cleanup arm instream()distinguishesGeneratorExit(consumer disconnect — drops state without yielding, since no consumer remains) from provider-side errors (yields defensivesubagent_endthen re-raises).For parallel dispatches of the same subagent type, each invocation has its own
tool_call_id. Prefertool_call_idfor routing tokens and fall back tosourceonly whentool_call_id == "".subagent_end.summaryis the subagent’s final response text — the same string the parent agent will see as thetasktool’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 |
|---|---|---|
|
| Base name for the per-project MongoDB store used for LangGraph checkpoints (AER mode only). Project scoping/discovery still applies unless overridden below. |
| (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). |
|
| Seconds for MongoDB checkpointer server selection before checkpoint IO fails. |
|
| Seconds for MongoDB checkpointer connection establishment. |
|
| 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.
Platform teaching docs (what Durable Workflow is, primitives, enablement, restrictions): Durable
Workflow. 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 <https://docs.langchain.com/oss/python/langgraph/graph-api#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
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