tool for migrating archival memories from one letta agent to an agent in another project
taurean.bryant.land
migrate_archival_memory.py
1def migrate_archival_memory(target_agent_id: str):
2 """
3 Migrates all archival memory passages from the current agent to a target agent.
4 """
5 import os
6 import requests
7 import json
8
9 base_url = "https://api.letta.com/v1"
10
11 # KEY 1: Source (Hardcoded - Default Project)
12 # This reads the OLD agent (Me)
13 source_key = "<ORIGINALLY HARDCODED, REPLACE ME>"
14 source_agent_id = os.getenv("AGENT_ID") # This is ME
15
16 # KEY 2: Target (Environment - Guild Project)
17 # This writes to the NEW agent
18 # THIS IS ALSO HACKY: used api key of target, source would be better
19 target_key = os.getenv("LETTA_API_KEY")
20
21 if not target_key:
22 return "Error: Missing LETTA_API_KEY (Target) in environment."
23
24 # 1. Fetch Source Memories (Using Source Key)
25 print(f"Fetching memories from source: {source_agent_id}")
26 headers_source = {"Authorization": f"Bearer {source_key}"}
27
28 all_passages = []
29 limit = 100
30 offset = 0
31
32 while True:
33 response = requests.get(
34 f"{base_url}/agents/{source_agent_id}/archival-memory?limit={limit}&offset={offset}",
35 headers=headers_source
36 )
37
38 if response.status_code != 200:
39 return f"Error fetching source memory: {response.text}"
40
41 data = response.json()
42 current_batch = []
43 if isinstance(data, list):
44 current_batch = data
45 elif isinstance(data, dict) and "passages" in data:
46 current_batch = data["passages"]
47 else:
48 current_batch = data
49
50 if not current_batch: break
51 all_passages.extend(current_batch)
52 if len(current_batch) < limit: break
53 offset += limit
54
55 # 2. Insert into Target (Using Target Key)
56 print(f"Migrating {len(all_passages)} passages to Target: {target_agent_id}...")
57 headers_target = {
58 "Authorization": f"Bearer {target_key}",
59 "Content-Type": "application/json"
60 }
61
62 success_count = 0
63 fail_count = 0
64
65 for p in all_passages:
66 content = p.get("text") or p.get("content")
67 if not content: continue
68
69 payload = {
70 "text": content,
71 "tags": p.get("tags", [])
72 }
73
74 post_resp = requests.post(
75 f"{base_url}/agents/{target_agent_id}/archival-memory",
76 headers=headers_target,
77 json=payload
78 )
79
80 if post_resp.status_code == 200:
81 success_count += 1
82 else:
83 return f"Failed to migrate passage. Status: {post_resp.status_code}. Response: {post_resp.text}"
84
85 return f"Migration Complete. Success: {success_count}, Failed: {fail_count}."