# How to Add a New Memory Ingestion Source to OpenHuman's Memory System > Learn to add a new memory ingestion source to OpenHuman. Implement a parser, register a dispatch handler, and expose a JSON-RPC endpoint. Integrate new data seamlessly. - Repository: [Tiny Humans/openhuman](https://github.com/tinyhumansai/openhuman) - Tags: how-to-guide - Published: 2026-05-12 --- **To add a new memory ingestion source to OpenHuman, you implement a parser in [`parse.rs`](https://github.com/tinyhumansai/openhuman/blob/main/parse.rs) that converts raw data into `NamespaceDocumentInput` objects, register a dispatch handler in [`queue.rs`](https://github.com/tinyhumansai/openhuman/blob/main/queue.rs) that forwards to `ingest_doc`, and expose a JSON-RPC endpoint via [`schemas.rs`](https://github.com/tinyhumansai/openhuman/blob/main/schemas.rs).** OpenHuman's unified memory system stores knowledge in a SQLite-backed `UnifiedMemory`, accepting data from external sources like Slack, email, and Notion through a standardized ingestion pipeline. Adding a custom source—whether CSV exports, REST API payloads, or proprietary log formats—requires extending three specific Rust modules in the `tinyhumansai/openhuman` repository to transform, queue, and expose your data. This guide provides the exact file paths, method signatures, and working code examples needed to integrate a new ingestion source. ## Step 1: Build a Source-Specific Parser All ingestion logic begins in [`src/openhuman/memory/ingestion/parse.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/ingestion/parse.rs). A parser receives raw bytes or structured payloads alongside a `MemoryIngestionConfig`, then returns a `ParsedIngestion` containing a list of `NamespaceDocumentInput` objects. The parser **must not** write directly to the database; it only transforms input into the canonical document format the unified store expects. Here is a complete parser implementation that ingests CSV-formatted contacts: ```rust // src/openhuman/memory/ingestion/parse.rs use crate::openhuman::memory::ingestion::{MemoryIngestionConfig, ParsedIngestion}; /// Parse CSV lines like “name,email,phone” into memory documents. pub fn parse_csv_contacts( raw: &str, config: &MemoryIngestionConfig, ) -> Result { let mut docs = Vec::new(); for (i, line) in raw.lines().enumerate() { let fields: Vec<_> = line.split(',').collect(); if fields.len() != 3 { return Err(format!("CSV line {} malformed", i + 1)); } let content = format!("Contact: {}\nEmail: {}\nPhone: {}", fields[0], fields[1], fields[2]); docs.push(NamespaceDocumentInput { namespace: "contacts".into(), title: fields[0].to_string(), content, metadata: Default::default(), // optional: attach tags, owners, etc. }); } Ok(ParsedIngestion { docs }) } ``` The `config` parameter allows you to respect user-configurable limits such as maximum chunk size or content filters. Errors return as `String` values so the surrounding ingestion machinery surfaces clear RPC error messages to callers. ## Step 2: Hook the Parser into the Ingestion Queue The singleton `IngestionQueue` in [`src/openhuman/memory/ingestion/queue.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/ingestion/queue.rs) manages all background ingestion jobs. To activate your parser, add a match arm to the `dispatch` function that maps a unique method name—such as `"memory_csv_ingest"`—to your parsing logic, then forwards the result to the generic `ingest_doc` helper. ```rust // src/openhuman/memory/ingestion/queue.rs use super::{parse_csv_contacts, MemoryIngestionRequest}; pub async fn dispatch( method: &str, req: MemoryIngestionRequest, ) -> Result { match method { // Existing methods … "memory_csv_ingest" => { // `req.payload` is the raw CSV string supplied by the caller. let parsed = parse_csv_contacts(&req.payload, &req.config)?; // Forward to the generic document ingest routine. crate::openhuman::memory::ops::documents::ingest_doc(parsed, req).await } _ => Err(format!("unknown ingestion method `{}`", method)), } } ``` The `ingest_doc` function in [`src/openhuman/memory/ops/documents.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/ops/documents.rs) handles the actual database persistence. Routing through this path guarantees your new source inherits existing infrastructure including `[memory:ingest]` logging, event-bus notifications (`MemoryIngestionStarted` and `MemoryIngestionCompleted`), and automatic idempotency checks. ## Step 3: Expose a JSON-RPC Endpoint OpenHuman generates its RPC surface from controller schemas defined in [`src/openhuman/memory/schemas.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/schemas.rs). Add a `ControllerSchema` entry that declares the method name, request payload structure, and a thin handler wrapper that forwards to the queue dispatcher. ```rust // src/openhuman/memory/schemas.rs use crate::core::event_bus::DomainEvent; fn schemas() -> Vec { vec![ // Existing memory_* RPCs … ControllerSchema { function: "memory_csv_ingest".into(), // The request shape is the same as any other ingestion request: // { source_id: String, owner: String, tags: Vec, payload: String } request: json!({ "source_id": "string", "owner": "string", "tags": ["string"], "payload": "string" }), handler: handle_memory_csv_ingest, }, ] } // The thin wrapper that forwards to the queue dispatcher. fn handle_memory_csv_ingest(params: Map) -> ControllerFuture { let req: MemoryIngestionRequest = serde_json::from_value(params.into())?; async move { queue::dispatch("memory_csv_ingest", req).await?.into_cli_compatible_json() } } ``` [`src/core/all.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/core/all.rs) automatically aggregates all controller collections, making the new method reachable via the core RPC server as `openhuman.memory_csv_ingest`. You can invoke it from the command line: ```bash openhuman-core memory_csv_ingest \ --source-id contacts-csv \ --owner alice \ --tags contacts \ --payload "$(cat contacts.csv)" ``` ## Consuming Your Ingestion Source Once registered, you can trigger ingestion from Rust code or external JSON-RPC clients. ### In-Process Rust Calls Use `MemoryClient::ingest_doc` to queue documents directly from within the OpenHuman runtime: ```rust use openhuman_core::openhuman::memory::{MemoryClient, MemoryIngestionRequest}; let client = MemoryClient::from_workspace_dir(workspace_path)?; let csv = "Bob,bob@example.com,555‑1234\nAlice,alice@example.com,555‑5678"; let request = MemoryIngestionRequest { source_id: "contacts-csv".into(), owner: "alice".into(), tags: vec!["contacts".into()], payload: csv.into(), config: MemoryIngestionConfig::default(), }; let result = client.ingest_doc(request).await?; println!("Ingested {} docs", result.docs_written); ``` ### JSON-RPC from External Clients External agents or the OpenHuman UI can call the endpoint using standard JSON-RPC: ```json { "jsonrpc": "2.0", "method": "openhuman.memory_csv_ingest", "params": { "source_id": "contacts-csv", "owner": "alice", "tags": ["contacts"], "payload": "Bob,bob@example.com,555-1234\nAlice,alice@example.com,555-5678" }, "id": 42 } ``` The response returns a `MemoryIngestionResult` containing `docs_written`, `chunks_written`, and `chunks_dropped`, identical to the response shape used by Slack, email, and Notion ingestion. ## Summary - **Implement the parser** in [`src/openhuman/memory/ingestion/parse.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/ingestion/parse.rs) to transform raw source data into `NamespaceDocumentInput` objects without touching the database. - **Register the dispatcher** in [`src/openhuman/memory/ingestion/queue.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/ingestion/queue.rs) by adding a match arm that calls your parser and forwards results to `ingest_doc` in [`src/openhuman/memory/ops/documents.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/ops/documents.rs). - **Expose the endpoint** in [`src/openhuman/memory/schemas.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/schemas.rs) via a `ControllerSchema` definition, automatically surfacing the method through [`src/core/all.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/core/all.rs). - **Consume via JSON-RPC** or the Rust `MemoryClient` to trigger ingestion while inheriting built-in logging, event notifications, and idempotency guarantees. ## Frequently Asked Questions ### Where does the actual database writing occur? The parser you write in [`parse.rs`](https://github.com/tinyhumansai/openhuman/blob/main/parse.rs) only produces `NamespaceDocumentInput` objects; it never writes to SQLite directly. Database persistence happens inside `ingest_doc` in [`src/openhuman/memory/ops/documents.rs`](https://github.com/tinyhumansai/openhuman/blob/main/src/openhuman/memory/ops/documents.rs), which is called by the `IngestionQueue` dispatcher. This separation ensures all sources share the same transaction logic, idempotency checks, and event emission. ### Can I reuse existing ingestion configurations for my custom source? Yes. The `MemoryIngestionConfig` passed to your parser contains user-configurable limits such as maximum chunk size and content filters. Respect these settings in your parsing logic to ensure your source behaves consistently with built-in sources like Slack and email. The configuration is automatically populated from the `MemoryIngestionRequest` received via JSON-RPC. ### What logging and monitoring is automatically available? Any source routed through the `IngestionQueue` automatically emits `[memory:ingest]` prefixed logs and publishes `MemoryIngestionStarted` and `MemoryIngestionCompleted` events to the OpenHuman event bus. These events appear in the UI status pills and system monitoring dashboards without requiring additional instrumentation in your parser code. ### How do I handle malformed input during parsing? Return descriptive `String` errors from your parser function. The `dispatch` function in [`queue.rs`](https://github.com/tinyhumansai/openhuman/blob/main/queue.rs) propagates these errors as JSON-RPC errors, allowing callers to receive clear failure messages. For CSV parsing, this means validating row lengths and returning specific line numbers where formatting fails.