From fded349fd47d3689bedbd52a9e9400b3e4658b40 Mon Sep 17 00:00:00 2001 From: jakerated-r Date: Sat, 13 Jun 2026 00:15:43 -0500 Subject: [PATCH 1/3] feat(sdk): add CrewAI tool adapter (EngramStoreTool/EngramSearchTool) --- engram/sdk/crewai.py | 187 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 187 insertions(+) create mode 100644 engram/sdk/crewai.py diff --git a/engram/sdk/crewai.py b/engram/sdk/crewai.py new file mode 100644 index 00000000..1b3d684c --- /dev/null +++ b/engram/sdk/crewai.py @@ -0,0 +1,187 @@ +""" +Engram — CrewAI tool adapter + +Exposes an Engram miner as CrewAI tools so any CrewAI agent can use the +decentralized subnet as long-term semantic memory: store facts during a run +and retrieve them later by meaning. + +Mirrors the lazy-import pattern of the LangChain and LlamaIndex adapters +(`engram/sdk/langchain.py`, `engram/sdk/llama_index.py`) so importing this +module never hard-requires CrewAI. + +Install: + pip install crewai engram-subnet + +Usage: + from crewai import Agent, Task, Crew + from engram.sdk.crewai import engram_memory_tools + + store_tool, search_tool = engram_memory_tools(miner_url="http://127.0.0.1:8091") + + researcher = Agent( + role="Researcher", + goal="Remember findings and recall them across tasks", + backstory="Uses Engram as durable cross-task memory.", + tools=[store_tool, search_tool], + ) + +You can also construct the tools directly: + + from engram.sdk.crewai import EngramStoreTool, EngramSearchTool + + store_tool = EngramStoreTool(miner_url="http://127.0.0.1:8091") + search_tool = EngramSearchTool(miner_url="http://127.0.0.1:8091", top_k=5) +""" + +from __future__ import annotations + +from typing import Any + +from engram.sdk.client import EngramClient +from engram.sdk.exceptions import EngramError + +try: + from crewai.tools import BaseTool + from pydantic import BaseModel, Field, PrivateAttr + + _CREWAI_AVAILABLE = True +except ImportError: # pragma: no cover - exercised only without CrewAI installed + _CREWAI_AVAILABLE = False + + # Minimal shims so the module imports cleanly and gives a helpful error + # at instantiation time, exactly like the LangChain/LlamaIndex adapters. + class BaseTool: # type: ignore[no-redef] + pass + + class BaseModel: # type: ignore[no-redef] + pass + + def Field(*args: Any, **kwargs: Any) -> Any: # type: ignore[no-redef] + return None + + def PrivateAttr(*args: Any, **kwargs: Any) -> Any: # type: ignore[no-redef] + return None + + +DEFAULT_MINER_URL = "http://127.0.0.1:8091" + + +def _require_crewai() -> None: + if not _CREWAI_AVAILABLE: + raise ImportError( + "crewai is required for the Engram CrewAI tools. " + "Install it with: pip install crewai" + ) + + +# ── Argument schemas ────────────────────────────────────────────────────────── + + +class _StoreArgs(BaseModel): + text: str = Field(..., description="The text/fact to store in Engram memory.") + source: str = Field( + default="crewai", + description="Optional tag recorded in the memory's metadata.", + ) + + +class _SearchArgs(BaseModel): + query: str = Field(..., description="What to recall, described in natural language.") + + +# ── Tools ─────────────────────────────────────────────────────────────────── + + +class EngramStoreTool(BaseTool): + """CrewAI tool that stores a piece of text in an Engram miner. + + The miner returns a content ID (CID); the same text+metadata is + content-addressed, so storing it twice yields the same CID. + """ + + name: str = "engram_store" + description: str = ( + "Store a fact or piece of text in long-term Engram memory so it can be " + "recalled later by meaning. Input the text to remember. " + "Returns the content ID (CID) of the stored memory." + ) + args_schema: type = _StoreArgs + + miner_url: str = DEFAULT_MINER_URL + timeout: float = 30.0 + + _client: EngramClient = PrivateAttr() + + def __init__(self, **data: Any) -> None: + _require_crewai() + super().__init__(**data) + self._client = EngramClient(miner_url=self.miner_url, timeout=self.timeout) + + def _run(self, text: str, source: str = "crewai") -> str: + try: + cid = self._client.ingest(text, metadata={"text": text, "source": source}) + return f"Stored memory. CID: {cid}" + except EngramError as exc: + return f"Could not store memory: {exc}" + + +class EngramSearchTool(BaseTool): + """CrewAI tool that retrieves the most relevant memories from an Engram miner.""" + + name: str = "engram_search" + description: str = ( + "Search long-term Engram memory for facts relevant to a natural-language " + "query. Returns the top matching memories with similarity scores." + ) + args_schema: type = _SearchArgs + + miner_url: str = DEFAULT_MINER_URL + timeout: float = 30.0 + top_k: int = 4 + + _client: EngramClient = PrivateAttr() + + def __init__(self, **data: Any) -> None: + _require_crewai() + super().__init__(**data) + self._client = EngramClient(miner_url=self.miner_url, timeout=self.timeout) + + def _run(self, query: str) -> str: + try: + results = self._client.query(query, top_k=self.top_k) + except EngramError as exc: + return f"Search failed: {exc}" + + if not results: + return "No relevant memories found." + + lines = [] + for r in results: + meta = dict(r.get("metadata") or {}) + content = meta.get("text", r.get("cid", "")) + score = float(r.get("score", 0.0)) + lines.append(f"[{score:.4f}] {content}") + return "\n".join(lines) + + +def engram_memory_tools( + miner_url: str = DEFAULT_MINER_URL, + timeout: float = 30.0, + top_k: int = 4, +) -> tuple[EngramStoreTool, EngramSearchTool]: + """Return a (store_tool, search_tool) pair wired to the same miner. + + Convenience for the common case of giving a CrewAI agent both + write and read access to Engram memory. + """ + _require_crewai() + store = EngramStoreTool(miner_url=miner_url, timeout=timeout) + search = EngramSearchTool(miner_url=miner_url, timeout=timeout, top_k=top_k) + return store, search + + +__all__ = [ + "EngramStoreTool", + "EngramSearchTool", + "engram_memory_tools", +] From cd7aaae955b21a288ff9b5f1e1c2f4d77f09ae01 Mon Sep 17 00:00:00 2001 From: jakerated-r Date: Sat, 13 Jun 2026 00:15:44 -0500 Subject: [PATCH 2/3] docs(examples): CrewAI agent memory example backed by Engram --- examples/crewai_agent_memory.py | 90 +++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) create mode 100644 examples/crewai_agent_memory.py diff --git a/examples/crewai_agent_memory.py b/examples/crewai_agent_memory.py new file mode 100644 index 00000000..86beb3a0 --- /dev/null +++ b/examples/crewai_agent_memory.py @@ -0,0 +1,90 @@ +""" +Example: give a CrewAI agent durable memory backed by Engram. + +This shows the SDK gap the LangChain and LlamaIndex adapters already fill for +their ecosystems: a CrewAI agent that can *store* findings during one task and +*recall* them by meaning in a later task, using an Engram miner as the backend. + +Prerequisites: + - A miner running locally: python -m neurons.miner (default http://127.0.0.1:8091) + - pip install crewai engram-subnet + +Run: + python examples/crewai_agent_memory.py + +If you just want to see the tools work without the full Crew/LLM loop, the +bottom of this file has a `--tools-only` path that calls the tools directly. +""" + +from __future__ import annotations + +import sys + +from engram.sdk.crewai import engram_memory_tools + +MINER_URL = "http://127.0.0.1:8091" + + +def tools_only_demo() -> None: + """Exercise the tools directly — no LLM or API key required.""" + store_tool, search_tool = engram_memory_tools(miner_url=MINER_URL) + + print("Storing two memories...") + print(" ", store_tool._run("Engram stores embeddings on a Bittensor subnet.")) + print(" ", store_tool._run("Retrieval uses the miner's HNSW index.")) + + print("\nRecalling 'how does Engram find similar items?'") + print(search_tool._run("how does Engram find similar items?")) + + +def full_crew_demo() -> None: + """Wire the tools into a real CrewAI Agent + Crew.""" + from crewai import Agent, Crew, Task + + store_tool, search_tool = engram_memory_tools(miner_url=MINER_URL, top_k=5) + + archivist = Agent( + role="Archivist", + goal="Persist important facts to Engram memory.", + backstory="You never forget — you write everything worth keeping to Engram.", + tools=[store_tool], + verbose=True, + ) + analyst = Agent( + role="Analyst", + goal="Answer questions using only what is in Engram memory.", + backstory="You recall stored facts by meaning before answering.", + tools=[search_tool], + verbose=True, + ) + + remember = Task( + description=( + "Store these facts in Engram memory: " + "(1) Engram is a decentralized vector database on Bittensor. " + "(2) Private namespaces use X25519 + sr25519 signed requests." + ), + expected_output="Confirmation that both facts were stored, with their CIDs.", + agent=archivist, + ) + recall = Task( + description="Using Engram memory, answer: how are private namespaces secured?", + expected_output="A short answer grounded in the retrieved memory.", + agent=analyst, + ) + + crew = Crew(agents=[archivist, analyst], tasks=[remember, recall], verbose=True) + result = crew.kickoff() + print(result) + + +if __name__ == "__main__": + if "--tools-only" in sys.argv: + tools_only_demo() + else: + try: + full_crew_demo() + except Exception as exc: # pragma: no cover - depends on local LLM/keys + print(f"Full crew run needs CrewAI + an LLM configured: {exc}") + print("Falling back to the tools-only demo (needs a running miner):\n") + tools_only_demo() From 663ce55f7dd6b51c347840f6d4be0eca6c200021 Mon Sep 17 00:00:00 2001 From: jakerated-r Date: Sat, 13 Jun 2026 00:15:45 -0500 Subject: [PATCH 3/3] test(sdk): offline-safe CrewAI adapter test --- scripts/test_crewai_sdk.py | 92 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 scripts/test_crewai_sdk.py diff --git a/scripts/test_crewai_sdk.py b/scripts/test_crewai_sdk.py new file mode 100644 index 00000000..cc8e77f1 --- /dev/null +++ b/scripts/test_crewai_sdk.py @@ -0,0 +1,92 @@ +""" +CrewAI adapter test — engram/sdk/crewai.py + +Offline-safe: requires neither a live miner nor an LLM. Mirrors the harness +style of scripts/test_sdk.py. + +Behavior covered: + - Module imports cleanly whether or not CrewAI is installed. + - Without CrewAI: constructing a tool raises a helpful ImportError. + - With CrewAI: tools construct, expose correct name/description/args_schema, + and degrade gracefully (return an error string, not raise) when the miner + is offline. + +Run: + python scripts/test_crewai_sdk.py +""" + +import sys + +sys.path.insert(0, ".") + +from engram.sdk import crewai as engram_crewai # noqa: E402 + +DEAD_URL = "http://127.0.0.1:19999" # nothing running here + +results: list[tuple[str, bool]] = [] +PASS = "✓" +FAIL = "✗" + + +def check(label: str, condition: bool) -> None: + print(f" {PASS if condition else FAIL} {label}") + results.append((label, bool(condition))) + + +def section(title: str) -> None: + print(f"\n[{title}]") + + +def main() -> int: + section("Import + public surface") + check("module imports", engram_crewai is not None) + check( + "exports the expected names", + set(engram_crewai.__all__) + == {"EngramStoreTool", "EngramSearchTool", "engram_memory_tools"}, + ) + + if not engram_crewai._CREWAI_AVAILABLE: + section("CrewAI not installed — graceful error path") + try: + engram_crewai.EngramStoreTool(miner_url=DEAD_URL) + check("EngramStoreTool() raises ImportError without CrewAI", False) + except ImportError: + check("EngramStoreTool() raises ImportError without CrewAI", True) + except Exception: + check("EngramStoreTool() raises ImportError without CrewAI", False) + else: + section("CrewAI installed — tool construction") + store = engram_crewai.EngramStoreTool(miner_url=DEAD_URL) + search = engram_crewai.EngramSearchTool(miner_url=DEAD_URL, top_k=3) + check("store tool name is 'engram_store'", store.name == "engram_store") + check("search tool name is 'engram_search'", search.name == "engram_search") + check("store has a non-empty description", bool(store.description)) + check("search args_schema is set", search.args_schema is not None) + + factory = engram_crewai.engram_memory_tools(miner_url=DEAD_URL) + check("engram_memory_tools() returns a pair", len(factory) == 2) + + section("Offline miner — graceful degradation (no exceptions)") + out_store = store._run("a fact to remember") + check("store._run returns a string", isinstance(out_store, str)) + check("store._run reports failure gracefully", "Could not store" in out_store) + + out_search = search._run("recall something") + check("search._run returns a string", isinstance(out_search, str)) + check("search._run reports failure gracefully", "Search failed" in out_search) + + failed = [label for label, ok in results if not ok] + print(f"\n{'='*52}") + print(f" {len(results) - len(failed)}/{len(results)} checks passed") + if failed: + print(" FAILED:") + for label in failed: + print(f" - {label}") + return 1 + print(" ALL GREEN") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())