Metadata-Version: 2.4
Name: metalins-drift
Version: 0.5.0
Summary: Drift Engine — continuous behavioral fingerprinting for AI agents. Detect model swaps, drift, and prompt injection without access to the model.
Project-URL: Homepage, https://metalins.ai
Project-URL: Documentation, https://metalins.ai/docs
Project-URL: Repository, https://github.com/Metalins/drift-engine-python
Author-email: Jose Hernandez <josemiguelhernandez16@gmail.com>
License: Apache-2.0
License-File: LICENSE
Keywords: agent-monitoring,agents,ai,behavioral-fingerprinting,drift-detection,drift-engine,metalins,prompt-injection
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27.0
Provides-Extra: behavioral
Requires-Dist: tiktoken>=0.6.0; extra == 'behavioral'
Provides-Extra: dev
Requires-Dist: langchain-core>=0.3.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: respx>=0.21.0; extra == 'dev'
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.3.0; extra == 'langchain'
Description-Content-Type: text/markdown

# metalins-drift (Python SDK)

The Python SDK for **Drift Engine** by [Metalins](https://metalins.ai) — the
open-source behavioral monitoring engine for AI agents in production.

Your critical AI agents in production are black boxes. Drift Engine watches them
continuously: same model, same behavior, same continuous process you deployed
— and raises alerts the moment that stops being true. Detect drift, silent
model swaps, successful prompt injection, RAG poisoning, and re-deploys that
shouldn't be there. Raw prompts and responses never leave your infrastructure:
the SDK hashes locally, only signed fingerprints reach your server. Three lines
of Python integrate any agent. **All behavioral scoring and comparison runs
server-side.**

> **Migrating from `metalins`?** This package was renamed from `metalins` to
> `metalins-drift` (Drift Engine is now a named product of the Metalins research
> lab). Update your install to `pip install metalins-drift` and your imports to
> `import metalins_drift`. The API is otherwise unchanged.

## Install

```bash
pip install metalins-drift
```

## Getting started — you run your own server

Drift Engine is **open source and self-hosted**. The SDK has **no default
server** — you point it at your own instance. The fastest way to get one is the
docker-compose stack in the repo root:

```bash
git clone https://github.com/Metalins/drift-engine
cd drift-engine
cp .env.example .env          # set your secrets
docker-compose up             # server on http://localhost:8000
```

Then tell the SDK where your server lives, either per call:

```python
agent = metalins_drift.Agent(
    api_key="ml_live_...",
    name="my-bot",
    base_url="http://localhost:8000",   # YOUR Drift Engine instance
)
```

…or via the environment (see `.env.example`):

```bash
export METALINS_BASE_URL=http://localhost:8000
export METALINS_API_KEY=ml_live_...
```

If neither `base_url` nor `METALINS_BASE_URL` is set, the SDK raises a
`ConfigurationError` instead of silently calling a shared endpoint.

## Quick start — the `Agent` facade

`Agent` is the one-import entry point for a long-lived agent. It registers the
agent on first run, persists its state so a restart resumes the same agent, and
runs a background loop that answers the server's verification checks on a
cadence — so verification keeps working whether or not the agent is busy.

```python
import metalins_drift

agent = metalins_drift.Agent(api_key="ml_live_...", name="my-customer-bot")
agent.start()                                  # background check loop on

# ... wherever the agent finishes a turn:
agent.log(input=user_message, output=agent_reply)

# Read status, or issue a signed identity proof for another party.
status = agent.get_status()
proof = agent.issue_proof(ttl_seconds=3600)

agent.stop()                                   # on shutdown
```

Or as a context manager, which starts and stops the loop for you:

```python
with metalins_drift.Agent(api_key="ml_live_...", name="my-customer-bot") as agent:
    agent.log(input=user_message, output=agent_reply)
```

The SDK hashes payloads locally — raw prompt and response text never leave your
process. The background loop does only hashing and HTTP; no model is involved.

## State persistence

`Agent` keeps its session — the agent id, its secret, and the running hash
chain — in a `StateStore` so a restart resumes the same agent. The default is a
local JSON file at `~/.metalins/<name>.json` with owner-only (`0600`)
permissions, zero config.

To keep the secret somewhere else (a database row, a secrets manager), pass any
object with `load() -> dict | None` and `save(dict) -> None`:

```python
agent = metalins_drift.Agent(api_key="ml_live_...", name="my-bot", store=my_store)
```

## LangChain

Attach the callback handler and every top-level chain or LLM call is logged
automatically — no explicit `agent.log(...)` in your turn code. Install the
extra: `pip install metalins-drift[langchain]`.

```python
from metalins_drift import Agent
from metalins_drift.integrations.langchain import MetalinsCallbackHandler

agent = Agent(api_key="ml_live_...", name="my-bot").start()
handler = MetalinsCallbackHandler(agent)

chain.invoke(user_input, config={"callbacks": [handler]})
```

## Lower-level: `Client` + `AgentSession`

`Agent` is built from two primitives you can also use directly. `Client` is a
thin wrapper with one method per developer-API endpoint; `AgentSession` holds
the per-agent hash chain needed to answer verification checks.

```python
ml = metalins_drift.Client(api_key="ml_live_...")

session = ml.start_session(name="my-bot", model="claude-sonnet")
session.log_event("user asked about pricing", "the agent's reply ...")

# Persist / rehydrate the session yourself.
saved = session.to_dict()
session = ml.attach_session(metalins_drift.AgentSession.from_dict(saved))
```

Every developer-API endpoint is also a direct method on `Client`
(`create_agent`, `log_event`, `answer_check`, `list_pending_checks`,
`list_agents`, `get_agent`, `issue_proof`, `revoke_agent`), each returning the
server's JSON response as a plain dict.

## Testing / sandbox mode

The SDK has no default server — it only talks to the `base_url` you give it (or
`METALINS_BASE_URL`). Every entry point accepts a `base_url` parameter, so you
can point your test suite at a mock or a throwaway local server without touching
any real instance.

### Option A — mock HTTP (fastest, no server)

Use [respx](https://lundberg.github.io/respx/) (or any `httpx`-compatible
mock) to intercept requests entirely. Zero real network calls.

```python
import respx
import metalins_drift
from httpx import Response

FAKE_BASE = "http://metalins.test"
FAKE_SECRET = "ab" * 32


@respx.mock
def test_my_agent_logs_an_event():
    # Stub the register + log endpoints.
    respx.post(f"{FAKE_BASE}/v1/agents").mock(
        return_value=Response(201, json={
            "agent_id": "agt_test",
            "agent_secret": FAKE_SECRET,
        })
    )
    respx.post(f"{FAKE_BASE}/v1/agents/agt_test/events").mock(
        return_value=Response(200, json={
            "event_count": 1,
            "pending_checks": [],
        })
    )

    agent = metalins_drift.Agent(
        api_key="ml_test_xxx",
        name="sandbox-bot",
        base_url=FAKE_BASE,
    )
    result = agent.log(input="hello", output="world")
    assert result["event_count"] == 1
```

Install respx once: `pip install respx`.

### Option B — ephemeral local server (full integration)

If you want real end-to-end coverage (real HTTP, real DB, real hash chain),
boot the Drift Engine server against a throwaway SQLite and tear it down after
the test session:

```python
import subprocess, sys, os, socket, time, httpx, pytest
import metalins_drift

def _free_port():
    s = socket.socket(); s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]; s.close(); return port

@pytest.fixture(scope="session")
def local_server(tmp_path_factory):
    """Boot a real Drift Engine server against a throwaway DB."""
    db = str(tmp_path_factory.mktemp("db") / "test.db")
    port = _free_port()
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "app.main:app",
         "--host", "127.0.0.1", "--port", str(port)],
        cwd="path/to/drift-engine/server",
        env={**os.environ,
             "METALINS_DB_URL": f"sqlite:///{db}",
             "METALINS_DISABLE_INPROC_SCHEDULER": "1"},
    )
    base = f"http://127.0.0.1:{port}"
    # Wait until ready.
    for _ in range(50):
        try:
            if httpx.get(base + "/", timeout=1).status_code == 200: break
        except Exception: pass
        time.sleep(0.3)
    yield base
    proc.terminate(); proc.wait()

@pytest.fixture(scope="session")
def sandbox_key(local_server):
    """Create a sandbox API key via the bypass-auth endpoint."""
    resp = httpx.post(
        local_server + "/internal/v1/customers/me/api-keys",
        headers={"X-Metalins-Test-Bypass": "..."},
        json={"name": "sandbox"},
    )
    return resp.json()["secret"]

def test_full_round_trip(local_server, sandbox_key):
    with metalins_drift.Agent(
        api_key=sandbox_key,
        name="sandbox-agent",
        base_url=local_server,
    ) as agent:
        result = agent.log(input="test input", output="test output")
        assert result["event_count"] == 1
        status = agent.get_status()
        assert status["agent_id"] == agent.agent_id
```

The `tests/e2e_core/` directory in the Drift Engine repo is the reference
implementation of this pattern: it boots the real ASGI app, seeds a sandbox
tenant, mints an API key via bypass-auth, and runs the full SDK ↔ developer-API
round-trip with no prod dependency.

### Which option to choose

| Scenario | Recommendation |
|---|---|
| Unit-testing your own code that wraps the SDK | Option A — respx mock |
| Integration-testing SDK + server together | Option B — ephemeral server |
| CI on a machine with the Drift Engine server source | Option B (see `tests/e2e_core/`) |
| Quick local smoke-test without server source | Option A |

Both options keep any real instance completely out of the picture — no
events leave your test run, no API key quota is consumed, and tests are
deterministic.

## License

Apache 2.0. See [LICENSE](LICENSE).
