Kioko

Pluggable, self-managing long-term memory for AI agents, for any framework, any language and any MCP host.

The name comes from kioku (記憶), Japanese for "memory": 記 (ki) "to record" and 憶 (oku) "to remember".

Kioko remembers what users tell your agents and hands the right memories back before every reply. That covers who users are, what they prefer, what they are working towards, what they decided, which procedures work and what happened. It manages that memory by itself.

• Fifteen kinds of memory. Profile, factual, semantic, episodic, working, conversational, goal, procedural, skill, strategy, decision, graph, KV-cache, parametric, and hybrid recall over all of them. Every kind works on every store and every surface.
• A memory graph that connects them. Every new memory is linked to the memories it relates to, with a type and the evidence: a decision serves a goal, a lesson refines a procedure, a fix resolves an incident. Recall follows the links and tells the agent how things connect. The algorithm is documented in full in graph memory.
• An intelligence that manages them. It routes each turn into the right kind with the right operation: progress on a goal, a decision that replaces an earlier one, a value that changed, "forget that", a ticket number for this session. It plans which kinds a question needs, and summarises, reflects, pins hot memories and tidies up in the background. It runs offline with rules and algorithms, or with any LLM, where the algorithms gate, validate and fall back.
• Efficient by design.
  • One model call per turn at most, and none for small talk.
  • Intent-based retrieval within a token budget: about 20% fewer tokens in the measured example.
  • Rolling session summaries.
  • A stable, cacheable prompt prefix that carries the memories needed most.
• Pluggable everywhere.
  • Stores: in-process, SQLite, Postgres + pgvector, Qdrant, or your own.
  • Embedders, extractors, rankers, processors, summarizers, framework adapters, and each part of the intelligence.
  • Plug in by class, by name, or with a pip-installable entry point.
• Three ways in, one behaviour.
  • attach() for any Python agent: Agno, LangChain, LangGraph, CrewAI, OpenAI Agents SDK, AutoGen, smolagents, plain functions.
  • A REST API with OpenAPI for any language or low-code tool.
  • An MCP server for GitHub Copilot, Claude, Cursor and Copilot Studio.
• No dependencies. The core, SQLite, the offline embedder, the intelligence, the REST/MCP server and the web console use only the Python 3.14 standard library.
• You can see and steer what it does. Every write, recall and decision is an event. The web console shows them live and lets you manage every user, session and memory. Webhooks push them elsewhere.

 any Python agent ──► attach() / with_memory / MemoryMiddleware ─┐
 any language     ──► REST  /v1   (OpenAPI)                      ├──► MemoryLayer ── 15 kinds of memory
 any MCP host     ──► MCP   /mcp  or stdio                       │      │  intelligence: route · plan · maintain
 other processes  ──► RemoteMemory / connect("http://...")       ┘      └─► store: SQLite · pgvector · Qdrant · yours
                                                                             events ──► console · webhooks

Five lines

import kioko

memory = kioko.intelligent("sqlite:///plant-memory.db")      # offline; kioko.intelligent(..., llm=ask) for a model

memory.record_turn("Hi, I'm Asha, a controls engineer at the Pune plant. I prefer short answers.",
                   "Nice to meet you, Asha!", user_id="asha", session_id="mon-shift")
memory.record_turn("My goal is to get line 3 OEE above 85% by December 31.",
                   "Noted - let's track it.", user_id="asha", session_id="mon-shift")
memory.record_turn("To reset the safety relay, switch off the main breaker, then press reset for 5 seconds, "
                   "then switch the breaker back on.", "Thanks, saved.", user_id="asha", session_id="mon-shift")

print(memory.recall("The safety relay tripped again - what do I do?", user_id="asha", session_id="tue-shift"))

<memory>
Long-term memory about the user. Use it to personalise your answer. ...

Instructions from the user:
- User prefers short answers.

About the user:
- User is a controls engineer.
- User works at the Pune plant.
- User's name is Asha.

The user's open goals:
- Get line 3 OEE above 85% by December 31 (medium, 0%, due 2026-12-31)

Procedures that may apply:
- reset the safety relay: 1) Switch off the main breaker; 2) Press reset for 5 seconds; 3) Switch the breaker back on
</memory>

Each sentence went to the kind where it belongs:

• name, role and employer became profile slots;
• the answer-style preference became a standing instruction;
• the goal became a goal with a due date;
• the how-to became a procedure.

The next day's question was recognised as a how-to. No model was called. The guide continues from here, through the agent loop, frameworks, stores, the service, MCP, the console and production.

Try it without writing code:

kioko demo                                  # scripted conversation in the terminal
kioko ui --db memory.db --intelligence      # web console at http://localhost:8765 (also serves /v1 and /mcp)

Documentation

Pick your way in

Install

Requires Python 3.14 or newer. On Windows, run the commands below through the launcher: py -3.14 -m ....

pip install kioko                 # the core: no third-party dependencies
pip install "kioko[pgvector]"     # + Postgres / pgvector (psycopg)
pip install "kioko[agno]"         # + the Agno adapter
pip install .                     # or from the source folder

Name. The package is kioko on PyPI. The standard romanization kioku is taken there by an unrelated cache library, and kioku-lite, kioku-agent-kit and kioku-mcp are other agent-memory projects. kioko keeps the meaning and stays distinct.

What's inside

Fifteen kinds of memory

Details and APIs: memory types. Runnable recipes: cookbook.

Every kind, on every surface

Each kind is reached the same way wherever your agent runs:

• Python, in-process or remote. The accessors above, on a MemoryLayer or on connect("http://...") (RemoteMemory has the same API).
• REST. POST /v1/kinds/{kind}/{operation}, with the method's arguments as JSON. GET /v1/kinds returns every kind's operations and counts.
• MCP. Dedicated tools for goals, decisions, lessons and the graph; save_memory for facts, preferences, instructions, knowledge and episodes.
• From conversation. With the intelligence on, manage(), record_turn(), POST /v1/manage, POST /v1/turns or MCP learn_from_conversation put each sentence into the right kind.

Whichever way a memory was written, it reaches the agent through recall() (REST POST /v1/recall, MCP get_memory_context).

All fifteen in one script (offline, no setup):

import kioko

memory = kioko.connect()            # in-process; connect("http://memory-service:8765") gives the same API remotely
user = "kumar"

memory.profile.set("location", "Pune", user_id=user)                                            # 1 profile
memory.facts.add("User's manager is Krutika", key="work.manager", user_id=user)                  # 2 factual
memory.semantic.add("OPC UA servers listen on TCP port 4840 by default", topic="iiot")           # 3 semantic (shared)
memory.episodes.record("Presented the maintenance demo", when="2026-10-20", user_id=user,         # 4 episodic
                       outcome="approved for a pilot")
memory.working("shift-1", user_id=user).set("ticket", "INC-4521")                                # 5 working
memory.conversation("shift-1", user_id=user).add_turn("The relay tripped", "Reset it at the panel")  # 6 conversational
goal = memory.goals.add("Ship Kioko 1.0", user_id=user, priority="high", due="2026-12-15")       # 7 goal
memory.procedures.add("Reset the safety relay", user_id=user,                                    # 8 procedural
                      steps=["Switch off the breaker", "Press reset for 5 seconds", "Switch it back on"])
memory.skills.add("summarize alarm log", description="Top recurring faults from a PLC alarm log",  # 9 skill
                  body="Summarize these alarms by fault code: {alarms}", user_id=user)
memory.strategies.reflect("debugging drives", "check the encoder cable first", "success", user_id=user)  # 10 strategy
memory.decisions.record("Vector store", chosen="pgvector", rationale="we already run Postgres",   # 11 decision
                        user_id=user)
memory.graph.add("Kumar", "reports_to", "Krutika", user_id=user)                                 # 12 graph
prefix = memory.kv.prefix(user_id=user)                                                          # 13 KV cache
rows = memory.parametric.export_dataset(user_id=user)                                           # 14 parametric
ctx = memory.recall("The relay tripped again, what do I do?", user_id=user, session_id="shift-1")  # 15 hybrid

print(ctx.text)

The same over REST (any language). Session-scoped kinds take a session_id:

H="Authorization: Bearer $KIOKO_API_KEY"
curl -s localhost:8765/v1/kinds/goals/add -H "$H" -H 'Content-Type: application/json' \
  -d '{"user_id": "kumar", "title": "Ship Kioko 1.0", "priority": "high", "due": "2026-12-15"}'
curl -s localhost:8765/v1/kinds/working/set -H "$H" -H 'Content-Type: application/json' \
  -d '{"user_id": "kumar", "session_id": "shift-1", "slot": "ticket", "value": "INC-4521"}'
curl -s localhost:8765/v1/kinds/decisions/record -H "$H" -H 'Content-Type: application/json' \
  -d '{"user_id": "kumar", "title": "Vector store", "chosen": "Qdrant", "rationale": "fast to set up"}'
curl -s localhost:8765/v1/kinds/decisions/supersede -H "$H" -H 'Content-Type: application/json' \
  -d '{"user_id": "kumar", "ref": "Vector store", "chosen": "pgvector", "rationale": "we already run Postgres"}'
curl -s "localhost:8765/v1/kinds?user_id=kumar" -H "$H"          # every kind's operations and counts

Writing each kind

Learning and recalling each kind

The intelligence

memory = kioko.intelligent()                         # or MemoryLayer(..., intelligence=True)
memory.manage("My goal is to pass the Azure AI Engineer exam by March 31.", user_id="kumar")
memory.manage("We decided to use Qdrant because it is fast.", user_id="kumar")

report = memory.manage("I'm about 60% done with the Azure exam prep, and we switched from Qdrant to pgvector "
                       "because we already run Postgres.", user_id="kumar", dry_run=True)
print(report.summary())
# [dry run] goal.progress "Pass the Azure AI Engineer exam by March 31" -> 60%
# [dry run] decision.supersede "use Qdrant" -> pgvector

Read intelligence for the operations, the LLM mode with validation and fallback, retrieval plans, maintenance (summaries, reflection, pinning hot memories, session wrap-up, consolidation) and the research behind it: Mem0, MIRIX, Generative Agents, MemGPT, MemOS, MemoryOS, LightMem and others.

The memory graph

memory.graph.add("Kioko", "uses", "pgvector", user_id="kumar")          # the entity graph
memory.manage("We decided to use pgvector because we already run Postgres.", user_id="kumar")
# linked: decision "use pgvector because we already run Postgres" serves goal "Ship the Kioko memory layer ..."
#         (why: via Kioko uses pgvector)

memory.graph.links(memory_id)          # typed links with their evidence, weight and author
memory.graph.related(memory_id)        # spreading activation: what is connected, with the chain of links
memory.graph.explain(a, b)             # why two memories are, or are not, linked
ctx = memory.recall("line 3 is faulting with E-217 again", user_id="kumar")
# ... brings the fix that worked last time, and a "How these memories connect:" section

Each new memory is anchored (entities, rare words), compared with candidates from an inverted index, nearest neighbours and its session, scored with a noisy-OR of shared terms, meaning, entity-graph bridges and time, typed from the pair of kinds and cue words, and stored on both memories. Feedback ("that worked") adds resolves links and strengthens what was used; unused links fade. Graph memory has the algorithm, with formulas, worked examples and measured cost.

Stores

docker compose -f docker-compose.stores.yml up -d starts Postgres with pgvector, Qdrant and a Kioko service. Read stores for the details.

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):

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

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])

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.

Run it as a service

export KIOKO_API_KEY="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"
kioko serve --db /data/memory.db --host 0.0.0.0 --port 8765 --intelligence

REST API            http://localhost:8765/v1   (OpenAPI: http://localhost:8765/v1/openapi.json)
MCP endpoint        http://localhost:8765/mcp

The agent loop is the same in every language: ask for memory before the LLM call, report the turn after it.

H='Authorization: Bearer '$KIOKO_API_KEY
curl -s localhost:8765/v1/recall -H "$H" -H 'Content-Type: application/json' \
  -d '{"user_id": "kumar", "session_id": "s1", "query": "Plan my weekend"}'     # -> {"text": "<memory>...", ...}
curl -s localhost:8765/v1/turns -H "$H" -H 'Content-Type: application/json' \
  -d '{"user_id": "kumar", "session_id": "s1", "user_message": "I moved to Pune", "assistant_message": "Noted!"}'

Import openapi.json (kioko openapi --server-url https://memory.example.com > openapi.json) into Postman, Swagger UI, Azure API Management or a Power Platform custom connector, or generate a typed client. Errors are JSON: {"error": "...", "code": "rejected"}, with 422 when a processor refused the content (for example a secret).

Security

One key grants access to every user's memories, which suits service-to-service calls. Put a gateway in front for per-user auth, and use HTTPS outside a private network.

RemoteMemory and connect()

from kioko import connect, attach

memory = connect("http://memory-service:8765")   # RemoteMemory; API key from $KIOKO_API_KEY
memory.goals.add("Ship v1", user_id="kumar")      # every kind works remotely
memory.manage("We switched to pgvector", user_id="kumar", dry_run=True)    # so does the intelligence
agent = attach(my_agent, memory)                  # every adapter works remotely

RemoteMemory has the MemoryLayer API and return types (MemoryRecord, SearchResult, WriteResult, MemoryContext, ManageReport, RecallPlan ...). Rejected writes return None, a rejected update raises MemoryRejected, and an unknown id gives None / KeyError. Other failures raise RemoteMemoryError (.status, 0 when the service can't be reached). Idempotent requests are retried. Redirects are not followed, so the key never leaves for another host. subscribe() polls the activity log. Calls that act on memories need a user_id.

connect(target) accepts:

kioko.intelligent(target, llm=None, ...) takes the same targets (except a service URL) and returns a self-managing layer. Anything with the same methods also satisfies the MemoryAPI protocol, so you can bridge another backend.

MCP server

The tools: get_memory_context (hybrid, planned recall), search_memory, list_memories, explore_entity, explore_memories (how memories connect), save_memory, learn_from_conversation, manage_goal, record_decision, record_lesson, relate_entities, update_memory and delete_memory. Start the server with --intelligence and learn_from_conversation also updates goals and decisions, keeps task details in working memory and forgets on request. Read-only tools are annotated, and --read-only exposes only those. The user whose memory the tools use is set by the server (--user, or the X-Kioko-User header over HTTP). The model can never choose it, so a prompt injection can't reach another user's memories. Protocol revisions 2024-11-05 to 2025-11-25 are supported.

Desktop hosts (stdio). For Claude Desktop (claude_desktop_config.json), Cursor (~/.cursor/mcp.json) and similar hosts:

{
  "mcpServers": {
    "kioko": { "command": "kioko", "args": ["mcp", "--user", "kumar"] }
  }
}

With no --db, memories go to ~/.kioko/memory.db. Use --url http://memory-service:8765 (plus KIOKO_API_KEY in env) to share a central service. For Claude Code: claude mcp add kioko -- kioko mcp --user kumar.

VS Code / GitHub Copilot (HTTP), in .vscode/mcp.json:

{
  "servers": {
    "kioko": {
      "type": "http",
      "url": "http://localhost:8765/mcp",
      "headers": { "X-Kioko-User": "kumar", "Authorization": "Bearer ${input:kioko-key}" }
    }
  },
  "inputs": [{ "type": "promptString", "id": "kioko-key", "description": "Kioko API key", "password": true }]
}

Copilot Studio and other remote clients. Use the Streamable HTTP endpoint https://<your-host>/mcp, with API-key authentication as Authorization: Bearer <key> or X-API-Key, served by kioko serve behind HTTPS. The server answers with JSON, keeps no sessions, and returns 405 to GET.

Webhooks

from kioko import WebhookSink

memory.subscribe(WebhookSink("https://hooks.example.com/memory", secret=os.environ["HOOK_SECRET"],
                             events=["memory.added", "memory.updated"]))   # prefixes like "memory." work too

Each event is POSTed as JSON from a background thread, so a slow or failing receiver never slows the agent. Each request carries X-Kioko-Event and X-Kioko-Delivery headers and, with a secret, X-Kioko-Signature: sha256=<HMAC>. Check the signature with kioko.sinks.verify_signature(secret, body, header). From the CLI: kioko serve --webhook URL --webhook-secret-env HOOK_SECRET.

Concepts

Extend it

from kioko import Adapter, Embedder, MemoryLayer, MemoryMiddleware, register

@register("embedder", "minilm")
class MiniLM(Embedder):
    def __init__(self):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer("all-MiniLM-L6-v2")
    def embed(self, texts):
        return self.model.encode(list(texts)).tolist()

@register("adapter", "my-framework")              # tried before the built-in adapters
class MyFrameworkAdapter(Adapter):
    def matches(self, target):
        return type(target).__module__.startswith("my_framework")
    def attach(self, target, memory, **options):
        hooks = MemoryMiddleware(memory, **options)
        target.before_turn(lambda t: t.add_context(str(hooks.before(t.text, user_id=t.user))))
        target.after_turn(lambda t: hooks.after(t.text, t.reply, user_id=t.user))
        return target

memory = MemoryLayer.from_config({          # or a path to a JSON file
    "store": {"type": "sqlite", "path": "memory.db"},
    "embedder": "minilm",                     # or {"type": "azure_openai", "deployment": "text-embedding-3-small"}
    "extractor": {"type": "llm", "llm": {"$import": "my_app.llm:complete"}},
    "processors": ["drop_secrets", "redact_pii"],
    "dedup_threshold": 0.92,
})

Ship components in their own package so pip install is all it takes. kioko components lists everything that's available:

[project.entry-points."kioko.embedders"]
minilm = "my_plugin.embedders:MiniLM"

[project.entry-points."kioko.adapters"]
my-framework = "my_plugin.adapter:MyFrameworkAdapter"

A custom store only needs five methods. See examples/04_custom_components.py for a JSON-file store in about 30 lines, and stores for the reusable contract tests. For a vector database (Azure AI Search, Elasticsearch, Milvus ...), implement vector_search, as the pgvector and Qdrant stores do. The extraction rules can be extended without a new extractor: RuleBasedExtractor(extra_rules=[Rule("ticket", r"^my ticket is (?P<id>[A-Z]+-\d+)$", template="User's open ticket is {id}", key="work.ticket", flags=0)]).

The console

kioko ui --db memory.db --intelligence                   # http://localhost:8765
kioko ui --store postgresql://... --agent my_agent:build # your database, your agent in the Playground
kioko ui --url http://memory-service:8765                # manage a running service from your laptop

More screenshots: playground · memories · kinds · memory graph · entity graph · recall · activity · integrate · integrate: MCP. How to reach a console on a server, and keep it safe: console.

The console also serves /v1 and /mcp. It is a local operator's tool. It binds to 127.0.0.1, its own pages and /api have no login (KIOKO_API_KEY protects /v1 and /mcp only), it checks Host and Origin, and it serves a strict Content-Security-Policy. Use kioko serve for anything shared.

CLI

kioko serve      [--db PATH | --store URL | --config JSON] [--intelligence | --intelligence-llm MOD:FN]
                     [--host] [--port] [--api-key-env NAME] [--cors-origin O ...] [--allowed-host NAME ...]
                     [--insecure] [--no-mcp] [--mcp-user U] [--mcp-read-only] [--webhook URL ...] [--webhook-secret-env NAME]
kioko ui         [--db | --store | --config | --url URL] [--intelligence] [--agent MODULE:FACTORY] [--open]
                     (+ the serve options except --insecure / --no-mcp)
kioko mcp        [--db | --store | --config | --url URL] [--intelligence] [--user U] [--agent-id A] [--read-only]
kioko recall Q   --user ID [--session S] [--budget N] [--plan] [--json]           [--db | --store | --url]
kioko manage T   --user ID [--reply R] [--session S] [--dry-run] [--json]         [--db | --store | --url]
kioko plan Q     --user ID [--json]                                               [--db | --store | --url]
kioko maintain   [--user ID] [--tasks reflect,pin,link,consolidate] [--json]      [--db | --store | --url]
kioko related ID [--depth N] [--json]                                            [--db | --store | --url]
kioko link A B   [--type T] [--why W] [--remove]                                 [--db | --store | --url]
kioko connect    [--user ID] [--rebuild] [--json]                                [--db | --store | --url]
kioko consolidate [--user ID] [--dry-run] [--json]                                [--db | --store | --url]
kioko kinds | list | search Q | add TEXT | context Q    (--user ID ...)           [--db | --store | --url]
kioko openapi    [--server-url URL]
kioko demo       [--db PATH] [--user ID]
kioko components

Deploy

A Dockerfile based on python:3.14-slim, a one-command local stack (compose.yml: service + console + Postgres/pgvector on http://127.0.0.1:8770, API key kioko-local-test-key) and docker-compose.stores.yml (Postgres + pgvector, Qdrant and the service) are included.

docker compose up --build -d --wait                           # local stack, uses examples/config/pgvector.json

docker build -t kioko .                                       # SQLite / Qdrant
docker build -t kioko --build-arg KIOKO_EXTRAS=pgvector .     # + Postgres/pgvector (psycopg)
docker run -p 8765:8765 -v kioko-data:/data -e KIOKO_API_KEY=... kioko \
  kioko serve --host 0.0.0.0 --db /data/memory.db --intelligence

export KIOKO_PG_PASSWORD=... KIOKO_API_KEY=...
docker compose -f docker-compose.stores.yml up -d             # Postgres + pgvector, Qdrant, the service

Production notes

• Embeddings. The default HashingEmbedder is lexical: it matches words, word pairs and character trigrams, not meaning, so "car" and "automobile" don't match. Use OpenAIEmbedder, AzureOpenAIEmbedder, AgnoEmbedder or FunctionEmbedder for semantic recall. After you change embedders, run memory.reembed() once. Tune HybridRanker(min_vector_only=...) to your embedder's cosine range.
• Extraction and management. The offline rules and intelligence cover common English phrasings: identity, location, work, likes and dislikes, instructions, goals and progress, decisions and changes, procedures, lessons, dated events, "remember ..." and "forget ...". For open-ended conversations and other languages, give the intelligence a model (kioko.intelligent(llm=...)); the rules stay as gate and fallback.
• Scale. Without vector_search, search scores every memory in the user's scope: about 5–30 ms for a few hundred memories, ~150 ms for 1,000. Past that, use the pgvector or Qdrant store. The service is a threaded standard-library server, which is enough for an internal agent fleet. For heavy traffic, run several instances behind a load balancer on Postgres, and lower context_cache_ttl.
• Secrets and PII. DropSecrets is on by default (passwords, API keys, tokens, private keys, card numbers). Add RedactPII to mask e-mails and phone numbers. Processors run on every write and every update, through any channel (Python, REST, MCP, tools, console).
• Languages. Storage, tokenization, the offline embedder and search work in any script, and tool and MCP output keeps non-ASCII text readable. The rule-based extraction only understands English, so use a model (LLMExtractor, or the intelligence with llm=) for other languages.
• Python 3.14 annotations. Annotations are evaluated lazily in 3.14 (PEP 649), so an agent function may annotate with a class defined later in the module. attach() / with_memory never force that evaluation: they read signatures with annotationlib.Format.FORWARDREF and copy the lazy __annotate__ function.
• Concurrency. SQLite runs in WAL mode with one serialized connection per process. That's fine for a service plus the console. For many writers, use Postgres + pgvector.

Development

python3.14 -m unittest discover -s tests -t . -v      # the full suite, standard library only (pytest also works)
python3.14 -m unittest tests.test_agno_contract -v    # with the real Agno installed
KIOKO_TEST_PG_DSN=postgresql://... KIOKO_TEST_QDRANT_URL=http://localhost:6333 \
  python3.14 -m unittest tests.test_stores tests.test_intelligence -v     # against real pgvector and Qdrant
python3.14 docs/build_cookbook.py                     # regenerate docs/cookbook.md from the recipes

CI (.github/workflows/tests.yml) runs on Python 3.14:

• on Linux and Windows, it compiles everything with warnings as errors (which catches 3.14 syntax warnings such as return inside finally, PEP 765), runs the suite from source, builds the wheel and sdist, and runs the suite again against the installed wheel;
• one job runs the store and intelligence contracts against real pgvector/pgvector:0.8.7-pg17 and qdrant/qdrant:v1.20.0 service containers;
• a separate, non-blocking job runs the Agno contract tests against the real agno package.

Releases are published by .github/workflows/release.yml. Bump the version in pyproject.toml and src/kioko/_version.py, then push a matching tag (git tag v0.6.1 && git push origin v0.6.1). The workflow checks that the tag and versions match, runs the suite, builds and checks the wheel and sdist, publishes to TestPyPI, installs the package back from there, publishes to PyPI (Trusted Publishing, no stored token) and creates a GitHub Release with the same files.

src/kioko/
  layer.py            MemoryLayer: write / learn / search / context / recall / manage / pins / events
  kinds/              the typed kinds: profile, facts, semantic, episodic, working, conversation, goals,
                      procedural, skills, strategies, decisions, graph, cache (KV), parametric, hybrid, maintenance
  intelligence/       routers (heuristic, LLM), planner, scheduler, operations, signals
  context.py          sections, rendering, token budgets             budget.py   token estimates
  connect.py          connect(): URL / path / config -> memory     protocol.py   MemoryAPI
  remote.py           RemoteMemory (HTTP client)                   sinks.py      WebhookSink
  service/            http.py (server, auth), rest.py (/v1), mcp.py (MCP), openapi.py
  integrations/       adapters.py (attach/detach, Adapter), generic.py (with_memory, MemoryMiddleware),
                      agno.py (AgnoMemory, AgnoEmbedder, export/import)
  stores/             memory, sqlite, pgvector, qdrant
  base.py             component interfaces        registry.py   names, config, entry points
  types.py            MemoryRecord, MemoryFilter, SearchResult, MemoryEvent, ...
  embedders/ extractors/ ranking.py processors.py summarizers.py tools.py
  ui/                 console app, demo agent, static web app
docs/                 guide, memory types, intelligence, cookbook (generated), stores, console, screenshots
examples/             quick start, Agno, any agent, custom components, console agent, service clients, cookbook/
tests/                unit + integration + contract tests (every store), fake Agno, fake Postgres, fake Qdrant

License

MIT. See LICENSE.