attach(): any agent, one line

from kioko import MemoryLayer, attach

memory = MemoryLayer.sqlite("memory.db")    # or kioko.intelligent("sqlite:///memory.db") to let it manage itself

# LangChain runnable or LangGraph graph: user and thread come from the config
chain = attach(prompt | llm, memory)
chain.invoke({"input": "I prefer short answers"},
             config={"configurable": {"user_id": "kumar", "thread_id": "t1"}})

# Agno agent: hooks + memory tools
agent = attach(Agent(model=OpenAIChat(id="gpt-4o-mini")), memory)
agent.run("What do you know about me?", user_id="kumar")

# CrewAI crew: say which input holds the user's text
crew = attach(my_crew, memory, input_key="topic", user_id="kumar")

# OpenAI Agents SDK, or anything else: wrap the call in a function
from agents import Agent, Runner
support = Agent(name="Support", instructions="You are a helpful assistant.")
ask = attach(lambda message: Runner.run_sync(support, message).final_output, memory)
ask("I prefer metric units", user_id="kumar")

attach(target, memory=None, *, adapter=None, **options) picks the first adapter whose matches() accepts the target, and returns what to use from now on (usually the same object):

Adapter	Targets	What it does
agno	Agno Agent / Team	Pre-hook recalls memories into the input, post-hook learns, plus save/search/update/delete_memory tools scoped to run_context.user_id.
method	Any object with invoke/ainvoke, stream/astream, run/arun, chat/achat, ask/aask, query/aquery, kickoff/kickoff_async: LangChain runnables, LangGraph graphs, CrewAI crews, your classes	Wraps those methods in place (same object, same type). Objects that refuse new attributes get a MemoryProxy. methods=[...] picks the methods.
callable	Functions and callable objects, sync, async or streaming	Returns a wrapped function (with_memory).
yours	Anything	Registered adapters and installed plug-ins are tried first (see Extend).

What is injected, and where. The memory block goes where the agent will read it, keeping the input's shape:

Input shape	Where the memory block goes
a string	in front of the text
a message list (OpenAI dicts, LangChain messages, ("human", "hi") tuples)	a system message after your own system messages
a dict with messages (LangGraph state, chat-completion requests)	the same, inside messages
a dict with input/query/question/prompt/message/...	in front of that field (input_key= picks another field)
a pydantic model or dataclass with such a field (LangChain prompt values too)	on a copy
one message with multi-part content (vision)	a leading text part

Who is talking. The user and session ids come from user_id= / session_id= / thread_id= arguments (removed before the call unless the target declares them), then a LangChain/LangGraph config={"configurable": {...}} (or config["metadata"]), then the payload dict itself. Without one, user_id= and session_id= given to attach() are the defaults. scope=lambda args, kwargs: ... decides first; it receives the call's original arguments, and if it returns None the rules above apply.

Replies. Replies are read from strings, {"output"|"answer"|"content": ...}, LangGraph state (the last assistant message), LangChain messages, Agno RunOutput, OpenAI Agents RunResult.final_output and CrewAI CrewOutput.raw. Streams (generators and async generators) pass through untouched and are recorded when they finish.

Nested calls. When a wrapped method calls another wrapped method of the same object (chat() → run(), or stream() → invoke(), even through a thread pool), memory is added once and the turn is recorded once. Async streams fetch memory off the event loop. If a stream fails, the agent's own error is raised, never a recording error.

detach(agent) undoes attach() and restores the original methods. Calling attach() again replaces the previous wrapping, so the newest options win. Attach once at startup and pass user_id with each call; don't re-attach per request. attach(agent, "http://memory:8765"), or attach(agent) with KIOKO_URL set, connects to a shared service instead of a local layer.

Checkpointed threads. For LangChain and LangGraph targets the injected system message carries a stable id (kioko-memory), so a checkpointed LangGraph thread replaces last turn's memory message instead of piling up a new one each turn. Memory blocks are never stored as history or learned from, even when they come back inside a later input.

Agno in detail

from kioko.integrations import AgnoMemory

mem = AgnoMemory(memory, inject="input", background=False)   # what attach() uses for Agno agents
agent = Agent(model=..., tools=mem.tools(), pre_hooks=[mem.pre_hook], post_hooks=[mem.post_hook])

Piece	What it does
pre_hook	Retrieves the user's instructions, profile facts and the memories relevant to the message, and puts the memory block in front of run_input.input_content.
post_hook	Stores the exchange as session history, extracts and writes new memories, then restores the original input in the stored run, so memory blocks don't pile up in Agno's history.
tools	save_memory, search_memory, update_memory, delete_memory. They declare run_context, so Agno injects the current run (hidden from the model) and the tools act on run_context.user_id.

Options: agent_id, inject="input" | "none", learn, tools, allow_update, allow_delete, context_limit, background (learn in a worker thread), restore_input. Also: LLMExtractor(llm=agno_agent), AgnoEmbedder(any_agno_embedder), and export_to_agno / import_from_agno to sync with Agno's built-in user memories. The adapter follows Agno's documented 3.x API (2.2.7+). Run python -m unittest tests.test_agno_contract -v once with Agno installed to check it against the real package.

Lower-level building blocks

from kioko import MemoryMiddleware, with_memory

@with_memory(memory)                                     # sync, async or streaming functions
def my_agent(message: str, memory_context: str = "") -> str:
    return call_llm(system=memory_context, user=message)

hooks = MemoryMiddleware(memory, agent_id="support-bot")  # inside a graph node, a task, your own loop
ctx = hooks.before(question, user_id="kumar", session_id="s1")   # str(ctx) is the prompt block
reply = run_my_agent(question, context=str(ctx))
hooks.after(question, reply, user_id="kumar", session_id="s1")   # history + learning

tools = memory.tools(user_id="kumar")                    # OpenAI / Azure OpenAI function calling
messages = memory.inject([{"role": "user", "content": question}], user_id="kumar")
response = client.chat.completions.create(model="gpt-4o-mini", messages=messages,
                                          tools=tools.openai_schemas())
for call in response.choices[0].message.tool_calls or []:
    output = tools.call(call.function.name, call.function.arguments)

tools.functions() returns plain functions with type hints and docstrings, ready to register as tools in LangChain (StructuredTool.from_function), CrewAI, smolagents, AutoGen or the OpenAI Agents SDK (function_tool). The tools only read and change the memories of the user they are bound to.