ebrain-sdk 0.3.0
Python SDK for E-Brain — Persistent External Memory for AI Agents
Requires Python: >=3.10
Keywords: agent, agentic, ai, context, langchain, llm, mcp, memory, openai, rag

# E-Brain Python SDK

Persistent external memory for AI agents. Give any LLM perfect recall across sessions.

## Installation

```bash
pip install ebrain-sdk
```

Or from source:

```bash
git clone https://github.com/thiefdroppers/ebrain.git
cd ebrain/sdks/python
pip install -e ".[dev]"
```

## Self-Service Quickstart (60 seconds)

```python
from ebrain import EBrain

# 1. Sign up — get your API key instantly (no auth required)
result = EBrain.signup("https://ebrain.riaax.com", email="dev@yourcompany.com")
api_key = result["api_key"]  # ebk_...

# 2. Connect
brain = EBrain("https://ebrain.riaax.com", token=api_key)

# 3. Verify the integration works
print(brain.verify())  # {"status": "verified", "message": "E-Brain integration successful!"}

# 4. Start using persistent memory
brain.remember("User prefers dark mode and vim keybindings", project="myapp")
results = brain.recall("user preferences")
print(results.results[0].content)  # "User prefers dark mode and vim keybindings"
```

## Manual Quickstart (existing token)

```python
from ebrain import EBrain

brain = EBrain("https://ebrain.riaax.com", token="your-token")
brain.remember("User prefers dark mode and vim keybindings", project="myapp")
results = brain.recall("user preferences")
print(results.results[0].content)  # "User prefers dark mode and vim keybindings"
```

## Async Usage

```python
from ebrain import EBrainAsync

async with EBrainAsync("https://ebrain.riaax.com", token="your-token") as brain:
    await brain.remember("SSH key is ed25519", project="infra")
    results = await brain.recall("SSH configuration")
```

## API Reference

### Core Memory

| Method | Description |
|--------|-------------|
| `remember(content, *, project, category, entities, confidence, session_id)` | Store a memory |
| `remember_alias(...)` | Store via `/v1/remember` alias |
| `recall(query, *, project, limit, tiers)` | Search memories |
| `recall_alias(...)` | Search via `/v1/recall` alias |
| `forget(memory_id)` | Delete a memory |
| `feedback(memory_id, *, score, comment)` | Rate a memory (+1/-1) |

### Sessions

| Method | Description |
|--------|-------------|
| `create_session(*, project, agent)` | Create a new session |
| `list_sessions()` | List all active sessions |
| `get_session(session_id)` | Get session details |
| `end_session(session_id)` | End and archive a session |
| `delete_session(session_id)` | Hard-delete a session |

### Context / VCM

| Method | Description |
|--------|-------------|
| `get_context(budget_tokens, *, project, session_id)` | Get optimal context for token budget |
| `report_context(*, session_id, context_tokens, max_tokens)` | Report context window state |
| `extract_context(content, *, project, session_id)` | Extract memories from text |
| `get_manifest(*, project)` | VCM knowledge manifest |
| `hot_compact(*, project, max_observations, target_tokens)` | Compact observations |
| `get_chain(*, project, limit, session_id)` | Query chain blocks |

### Agentic

| Method | Description |
|--------|-------------|
| `augment(prompt, *, project, conversation_history, max_tokens)` | Inject memories into prompt |
| `consult(query, *, project, limit, tiers)` | Explicit memory lookup |
| `agentic_feedback(memory_id, *, useful, context)` | Report memory usefulness |
| `detect_uncertainty(text, *, threshold)` | Find hedging in AI text |

### Pins

| Method | Description |
|--------|-------------|
| `create_pin(content, *, project, category, tags, cue_keywords, priority)` | Create always-inject memory |
| `list_pins(*, project)` | List pinned memories |
| `update_pin(pin_id, *, content, project, ...)` | Update a pin |
| `delete_pin(pin_id)` | Delete a pin |

### Webhooks

| Method | Description |
|--------|-------------|
| `send_event(event_type, data, *, project, session_id)` | Generic event |
| `send_tool_event(tool_name, tool_input, *, tool_output, ...)` | Tool use event |
| `send_chat(messages, *, project, session_id)` | Chat messages |
| `send_batch(events, *, project, session_id)` | Bulk events |

### MCP (Model Context Protocol)

| Method | Description |
|--------|-------------|
| `mcp_info()` | Server info |
| `mcp_list_tools()` | List available tools |
| `mcp_call_tool(name, arguments)` | Call a tool |
| `mcp_list_resources()` | List resources |
| `mcp_read_resource(uri)` | Read a resource |
| `mcp_rpc(method, *, params, request_id)` | JSON-RPC 2.0 |

### Proxy (OpenAI-Compatible)

| Method | Description |
|--------|-------------|
| `chat(prompt, *, model, system, temperature, ...)` | Chat with memory injection |
| `chat_completions(messages, *, model, stream, ...)` | Full OpenAI-compatible endpoint |
| `proxy_status()` | Proxy health and stats |

### Onboard

| Method | Description |
|--------|-------------|
| `get_onboard_config()` | Auto-onboarding configuration |
| `detect_agent()` | Detect agent type |

### Self-Service

| Method | Description |
|--------|-------------|
| `EBrain.signup(url, email, *, company, plan)` | **Static** — Create account, get API key |
| `verify()` | Test full integration pipeline |

### System

| Method | Description |
|--------|-------------|
| `health()` | Health check |
| `metrics()` | System metrics |
| `stats()` | Comprehensive statistics |
| `debug_tiers()` | Debug tier information |
| `gc()` | Trigger garbage collection |
| `consolidate()` | Trigger memory consolidation |

## Streaming Chat

```python
brain = EBrain("https://ebrain.riaax.com", token="your-token")

# Stream responses token by token
for chunk in brain.chat("Tell me about quantum computing", stream=True):
    if chunk.choices and chunk.choices[0].delta:
        content = chunk.choices[0].delta.content
        if content:
            print(content, end="", flush=True)
```

Async streaming:

```python
async with EBrainAsync("https://ebrain.riaax.com", token="tok") as brain:
    stream = await brain.chat("Tell me a story", stream=True)
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta:
            print(chunk.choices[0].delta.content or "", end="")
```

## Error Handling

```python
from ebrain import (
    EBrain,
    EBrainError,           # Base exception
    EBrainAuthError,       # 401
    EBrainNotFoundError,   # 404
    EBrainRateLimitError,  # 429 (has .retry_after)
    EBrainServerError,     # 5xx
    EBrainConnectionError, # Network errors
    EBrainValidationError, # 422
)

brain = EBrain("https://ebrain.riaax.com", token="your-token")

try:
    brain.forget("nonexistent-id")
except EBrainNotFoundError:
    print("Memory not found")
except EBrainRateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except EBrainError as e:
    print(f"E-Brain error: {e.message} (HTTP {e.status_code})")
```

## Retry Configuration

The client automatically retries on transient errors (429, 500, 502, 503) with exponential backoff:

```python
brain = EBrain(
    "https://ebrain.riaax.com",
    token="your-token",
    retry_count=5,      # Max 5 retries (default: 3)
    retry_delay=0.5,    # Start with 0.5s delay, doubles each retry
    timeout=60.0,       # 60s request timeout
)
```

## Framework Integration

### LangChain

```python
from langchain.memory import ConversationBufferMemory
from ebrain import EBrain

brain = EBrain("https://ebrain.riaax.com", token="tok")

class EBrainLangChainMemory(ConversationBufferMemory):
    def save_context(self, inputs, outputs):
        super().save_context(inputs, outputs)
        brain.remember(
            f"User: {inputs['input']}\nAssistant: {outputs['output']}",
            project="langchain-app",
            category="conversation",
        )

    def load_memory_variables(self, inputs):
        # Augment with E-Brain memories
        result = brain.augment(inputs.get("input", ""), project="langchain-app")
        return {"history": result.augmented_prompt}
```

### CrewAI

```python
from crewai import Agent, Task, Crew
from ebrain import EBrain

brain = EBrain("https://ebrain.riaax.com", token="tok")

def memory_tool(query: str) -> str:
    """Search E-Brain memory for relevant context."""
    results = brain.recall(query, limit=5)
    return "\n".join(r.content for r in results.results)

agent = Agent(
    role="Analyst",
    goal="Analyze data with full memory recall",
    tools=[memory_tool],
)
```

### Direct OpenAI Client Replacement

```python
# Instead of:
# from openai import OpenAI
# client = OpenAI()

from ebrain import EBrain
brain = EBrain("https://ebrain.riaax.com", token="tok")

# Same interface, but with automatic memory injection
response = brain.chat_completions(
    messages=[{"role": "user", "content": "What did we discuss yesterday?"}],
    model="gpt-4",
)
print(response.choices[0].message.content)
```

## Development

```bash
# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run linter
ruff check .

# Type check
mypy ebrain/
```

## License

Apache-2.0