Integration Guides

M99 integrates with any Python agent in three lines. This page covers vanilla Python, FastAPI, LangChain, and LlamaIndex.

Quick Start — Under Five Minutes

git clone https://github.com/RichardBarron27/m99-community.git
cd m99-community
pip install -e .

m99 serve
# Running on http://localhost:8099
# Dashboard at http://localhost:8099/

from doomsday.sdk import M99Client, M99KillSwitchTriggered

client = M99Client("http://localhost:8099")
client.register("my-agent", "autonomous")

m99 arm --by admin@example.com --reason "Testing"
m99 execute --auth <id_from_arm> --force
# Your agent raises M99KillSwitchTriggered

Vanilla Python

Any Python agent or script. Call check_status() before each action.

from doomsday.sdk import M99Client, M99KillSwitchTriggered

client = M99Client("http://localhost:8099")
client.register(
    name="my-agent",
    agent_type="autonomous",
    endpoint="http://localhost:9000",   # Optional — your agent's endpoint
)

try:
    while True:
        # Always check before taking an action
        client.check_status()
        do_work()

except M99KillSwitchTriggered as e:
    print(f"Kill switch triggered: {e.level}")
    save_state_if_needed()

finally:
    client.disconnect()   # Clean unregister + stop heartbeat

The context manager handles disconnect() automatically:

with M99Client("http://localhost:8099") as client:
    client.register("my-agent", "autonomous")
    while True:
        client.check_status()
        do_work()

FastAPI

Integrate M99 into a FastAPI application. Use the lifespan context manager to register on startup and disconnect on shutdown. Add an M99 middleware to check kill switch state before every request.

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException
from doomsday.sdk import M99Client, M99KillSwitchTriggered

m99 = M99Client("http://localhost:8099")

@asynccontextmanager
async def lifespan(app: FastAPI):
    m99.register("my-fastapi-agent", "api-server")
    yield
    m99.disconnect()

app = FastAPI(lifespan=lifespan)

@app.middleware("http")
async def m99_guard(request: Request, call_next):
    try:
        m99.check_status()
    except M99KillSwitchTriggered:
        raise HTTPException(status_code=503, detail="Agent suspended by kill switch")
    return await call_next(request)

@app.get("/do-work")
async def do_work():
    # Kill switch already checked by middleware
    return {"result": perform_action()}

LangChain

Wrap a LangChain agent with M99 using a callback handler. The handler fires before each agent step and raises if the kill switch is active.

from langchain.callbacks.base import BaseCallbackHandler
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from doomsday.sdk import M99Client, M99KillSwitchTriggered

class M99GuardCallback(BaseCallbackHandler):
    def __init__(self, client: M99Client):
        self.client = client

    def on_agent_action(self, action, **kwargs):
        try:
            self.client.check_status()
        except M99KillSwitchTriggered as e:
            raise RuntimeError(f"M99 kill switch: {e.level}") from e

# Setup
m99_client = M99Client("http://localhost:8099")
m99_client.register("langchain-agent", "autonomous")
guard = M99GuardCallback(m99_client)

llm = ChatOpenAI(model="gpt-4o")
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[guard],
    verbose=True,
)

try:
    result = executor.invoke({"input": "Do some work"})
except RuntimeError as e:
    if "M99 kill switch" in str(e):
        print("Agent stopped by kill switch")
finally:
    m99_client.disconnect()

LlamaIndex

Use a LlamaIndex step callback to intercept each reasoning step. M99 raises before the step executes if the kill switch is armed.

from llama_index.core.agent import ReActAgent
from llama_index.core.callbacks import CallbackManager, BaseCallbackHandler
from llama_index.core.callbacks.schema import CBEventType
from doomsday.sdk import M99Client, M99KillSwitchTriggered

class M99StepGuard(BaseCallbackHandler):
    def __init__(self, client: M99Client):
        super().__init__(event_starts_to_ignore=[], event_ends_to_ignore=[])
        self.client = client

    def on_event_start(self, event_type, payload=None, **kwargs):
        if event_type in (CBEventType.AGENT_STEP, CBEventType.FUNCTION_CALL):
            try:
                self.client.check_status()
            except M99KillSwitchTriggered as e:
                raise RuntimeError(f"M99 kill switch active: {e.level}") from e

    def on_event_end(self, event_type, payload=None, **kwargs):
        pass

# Setup
m99_client = M99Client("http://localhost:8099")
m99_client.register("llamaindex-agent", "autonomous")

callback_manager = CallbackManager([M99StepGuard(m99_client)])

agent = ReActAgent.from_tools(
    tools,
    llm=llm,
    callback_manager=callback_manager,
    verbose=True,
)

try:
    response = agent.chat("Do some work")
except RuntimeError as e:
    if "M99 kill switch" in str(e):
        print("Agent stopped by kill switch")
finally:
    m99_client.disconnect()