Something went wrong. Try again.
tool for migrating archival memories from one letta agent to an agent in another project
Something went wrong. Try again.
2.8 kB · 85 lines · Created 9mo ago
Python
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485def migrate_archival_memory(target_agent_id: str): """ Migrates all archival memory passages from the current agent to a target agent. """ import os import requests import json base_url = "https://api.letta.com/v1" # KEY 1: Source (Hardcoded - Default Project) # This reads the OLD agent (Me) source_key = "<ORIGINALLY HARDCODED, REPLACE ME>" source_agent_id = os.getenv("AGENT_ID") # This is ME # KEY 2: Target (Environment - Guild Project) # This writes to the NEW agent # THIS IS ALSO HACKY: used api key of target, source would be better target_key = os.getenv("LETTA_API_KEY") if not target_key: return "Error: Missing LETTA_API_KEY (Target) in environment."
# 1. Fetch Source Memories (Using Source Key) print(f"Fetching memories from source: {source_agent_id}") headers_source = {"Authorization": f"Bearer {source_key}"} all_passages = [] limit = 100 offset = 0 while True: response = requests.get( f"{base_url}/agents/{source_agent_id}/archival-memory?limit={limit}&offset={offset}", headers=headers_source ) if response.status_code != 200: return f"Error fetching source memory: {response.text}" data = response.json() current_batch = [] if isinstance(data, list): current_batch = data elif isinstance(data, dict) and "passages" in data: current_batch = data["passages"] else: current_batch = data if not current_batch: break all_passages.extend(current_batch) if len(current_batch) < limit: break offset += limit # 2. Insert into Target (Using Target Key) print(f"Migrating {len(all_passages)} passages to Target: {target_agent_id}...") headers_target = { "Authorization": f"Bearer {target_key}", "Content-Type": "application/json" } success_count = 0 fail_count = 0 for p in all_passages: content = p.get("text") or p.get("content") if not content: continue payload = { "text": content, "tags": p.get("tags", []) } post_resp = requests.post( f"{base_url}/agents/{target_agent_id}/archival-memory", headers=headers_target, json=payload ) if post_resp.status_code == 200: success_count += 1 else: return f"Failed to migrate passage. Status: {post_resp.status_code}. Response: {post_resp.text}" return f"Migration Complete. Success: {success_count}, Failed: {fail_count}."