metalins-drift 0.5.0

Drift Engine — continuous behavioral fingerprinting for AI agents. Detect model swaps, drift, and prompt injection without access to the model.

metalins-drift (Python SDK)

The Python SDK for Drift Engine by Metalins — 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

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:

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:

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

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.

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:

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:

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

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.

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 (or any httpx-compatible mock) to intercept requests entirely. Zero real network calls.

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:

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.