Stateful Agentic Engineering Copilots with Letta: Core Memory, Shared State, Semantic Retrieval, Custom Tools, Streaming, and Portable Agent Workflows

In this tutorial, we build a stateful engineering copilot with Letta that demonstrates how agentic memory extends beyond conventional prompt-based conversations. We begin by creating inspectable core memory blocks that the agent can autonomously update, then expose those blocks for programmatic state synchronization and shared access across multiple agents. We further extend the architecture with archival vector memory for semantic retrieval, a custom server-side tool for deterministic risk estimation, streaming for observing agent execution, server-side message history for persistent conversations, and .af export for portability. Together, these components demonstrate a practical memory architecture in which persistent state, retrieval, tools, multi-agent collaboration, and execution traces work as an integrated agentic system. Copy Code Copied Use a different Browser import subprocess, sys subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", "letta-client"], check=True) import os, json, time, textwrap from getpass import getpass from letta_client import Letta def flex(*attempts, label=""): err = None for fn in attempts: try: return fn() except Exception as e: err = e raise RuntimeError(f"All call variants failed for {label or 'call'}: {err}") from err API_KEY = os.environ.get("LETTA_API_KEY") or getpass("Letta API key (platform.letta.com/api-keys): ") BASE_URL = os.environ.get("LETTA_BASE_URL", "").strip() client = flex( lambda: Letta(base_url=BASE_URL, api_key=API_KEY) if BASE_URL else Letta(api_key=API_KEY), lambda: Letta(base_url=BASE_URL, token=API_KEY) if BASE_URL else Letta(token=API_KEY), label="client init", ) def hr(title): print("\n" + "=" * 92 + f"\n {title}\n" + "=" * 92) def as_text(content): if isinstance(content, str): return content if isinstance(content, list): return "".join(getattr(p, "text", "") or "" for p in content) return str(content or "") def show(response, hide_reasoning=False): for m in getattr(response, "messages", response) or []: kind = getattr(m, "message_type", None) or getattr(m, "type", "?") if kind == "reasoning_message" and not hide_reasoning: print(" reasoning :", textwrap.shorten(getattr(m, "reasoning", "") or "", 220)) elif kind == "tool_call_message": tc = getattr(m, "tool_call", None) print(" tool call :", getattr(tc, "name", "?"), textwrap.shorten(str(getattr(tc, "arguments", "")), 200)) elif kind == "tool_return_message": print(" tool ret :", textwrap.shorten(str(getattr(m, "tool_return", "")), 260)) elif kind == "assistant_message": print(" assistant :", as_text(getattr(m, "content", ""))) elif kind == "approval_request_message": print(" approval : agent is waiting for you to approve a tool call") u = getattr(response, "usage", None) if u: print(f" ↳ steps={getattr(u, 'step_count', '?')} tokens={getattr(u, 'total_tokens', '?')}") def chat(agent_id, text, max_steps=12, quiet=False): r = flex( lambda: client.agents.messages.create(agent_id=agent_id, input=text, max_steps=max_steps), lambda: client.agents.messages.create(agent_id=agent_id, input=text), lambda: client.agents.messages.create(agent_id=agent_id, messages=[{"role": "user", "content": text}]), label="messages.create", ) if not quiet: print(f"\n user : {text}") show(r) return r PREFERRED = ["openai/gpt-4.1", "openai/gpt-4o-mini", "anthropic/claude-sonnet-4-20250514"] try: handles = [getattr(m, "handle", None) for m in client.models.list()] MODEL = next((h for h in PREFERRED if h in handles), None) or handles[0] except Exception: MODEL = PREFERRED[0] print(f" connected · model handle = {MODEL}") hr("1. Create a stateful agent with custom memory blocks") def make_agent(**kw): return flex( lambda: client.agents.create(model=MODEL, embedding="openai/text-embedding-3-small", **kw), lambda: client.agents.create(model=MODEL, **kw), label="agents.create", ) copilot = make_agent( name=f"copilot-{int(time.time())}", description="Engineering copilot with inspectable long-term memory", tags=["tutorial", "copilot"], memory_blocks=[ {"label": "persona", "value": "I am Mnemo, an engineering copilot. I am terse, concrete, and I never " "invent facts — if I don't know something I say so and then go look it up.", "limit": 3000}, {"label": "human", "description": "Durable facts about the engineer I work with: name, role, stack, " "preferences, working style. Update this whenever I learn something stable.", "value": "Unknown so far.", "limit": 3000}, {"label": "project_state", "description": "Live state of the project we are shipping: milestones, blockers, " "decisions and their rationale. Keep it dense and current; delete stale lines.", "value": "(empty)", "limit": 5000}, {"label": "lessons", "description": "Hard-won lessons and mistakes to never repeat. Append-only, one line each.", "value": "(empty)", "limit": 2000}, ], ) print("agent id:", copilot.id) def get_block(agent_id, label): return flex( lambda: client.agents.blocks.retrieve(agent_id=agent_id, block_label=label), lambda: client.agents.blocks.retrieve(agent_id, label), label="agents.blocks.retrieve", ) def set_block(agent_id, label, value): return flex( lambda: client.agents.blocks.update(agent_id=agent_id, block_label=label, value=value), lambda: client.agents.blocks.modify(agent_id=agent_id, block_label=label, value=value), lambda: client.agents.blocks.update(agent_id, label, value=value), label="agents.blocks.update", ) def dump_memory(agent_id, title="core memory"): blocks = flex(lambda: client.agents.blocks.list(agent_id=agent_id), lambda: client.agents.blocks.list(agent_id), label="agents.blocks.list") print(f"\n──── {title} ────") for b in blocks: ro = " [read-only]" if getattr(b, "read_only", False) else "" print(f"▸ {b.label}{ro} ({len(b.value or '')}/{getattr(b, 'limit', '?')} chars)") print(textwrap.indent(textwrap.fill(b.value or "", 100), " ")) return blocks dump_memory(copilot.id, "core memory at birth") We set up the Letta client, install the SDK, and add compatibility helpers that handle API variations across Letta versions. We dynamically select an available model and create a stateful engineering copilot with structured core memory blocks. We then inspect the initial memory state to establish the agent’s persistent context. Copy Code Copied Use a different Browser hr("2. Self-editing memory — talk once, then read the block back") chat(copilot.id, "Hi, I'm Priya, staff engineer at Northwind. We're shipping 'Halo', a Rust ingestion " "service, on Nov 14. I hate long explanations — give me bullets or code, never essays. " "Current blocker: the Kafka consumer rebalances every ~4 minutes under load.") chat(copilot.id, "Update: we traced the rebalance to a 5s max.poll.interval.ms with slow batch commits. " "Decision: move commits off the poll loop. Also remember we already tried bumping " "session.timeout.ms and it did nothing — don't suggest that again.") dump_memory(copilot.id, "core memory after 2 turns (note what moved where)") hr("3. Your code writes into the context window") live = get_block(copilot.id, "project_state") set_block(copilot.id, "project_state", (live.value or "") + "\n[CI 09:41Z] main is RED: integration/kafka_rebalance_test failing (flaky x3).") chat(copilot.id, "Anything I should know before I start the day?") hr("4. Shared memory between two agents") policies = client.blocks.create( label="policies", description="Non-negotiable engineering policies. Read-only: obey, never edit.", value="1. No direct pushes to main.\n2. Every fix ships with a regression test.\n" "3. Customer data never leaves EU regions.", limit=2000, read_only=True, ) scratchpad = client.blocks.create( label="team_scratchpad", description="Shared working notes between the copilot and the reviewer agent. " "Both agents read and write here; keep entries prefixed with your name.", value="(empty)", limit=4000, ) for bid in (policies.id, scratchpad.id): flex(lambda: client.agents.blocks.attach(agent_id=copilot.id, block_id=bid), lambda: client.agents.blocks.attach(copilot.id, bid), label="agen