By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Build a Company-Wide Multi-Agent Harness with LangGraph

In 2026, Y Combinator open-sourced QM (MIT license, GitHub org yc-software) and the one-word summary is multiplayer. QM is not an agent that answers questions for a single developer; it is a "multiplayer" multi-agent harness designed for a whole company. YC itself runs accounting, legal, events, and engineering work on it — including developing QM with QM. It is cloud-first, ships native Slack and web interfaces, and is deliberately model-agnostic: the same orchestration loop runs on Pi, OpenCode, Codex, or Claude Code, whichever model happens to be best at a given task on a given day. That is the "harness" framing: the orchestration is the product, the model is a swappable worker.

The mental shift matters. A single-user agent needs context and tools. A company harness needs roles, routing, permissions, approvals, and governance — because the moment agents can touch accounting, legal, and engineering simultaneously, "the model" is no longer the risk surface. The routing and approval layers are. This dispatch builds company-harness, a LangGraph workflow that generalizes QM's architecture: a task intake router (Slack, web, email), a role router that dispatches to the right agent on the right model backend through a model-agnostic adapter, an approval board for cross-role decisions, a unified observability and audit layer, and a self-improvement loop where agents can file tasks back into the intake queue.

The architecture: roles, not agents

The first design decision is to model roles as the durable entities and agents as ephemeral workers. Engineering, Finance, Legal, and Ops each have a role manifest: what skills they hold, what they may touch, what they must not touch, and what they must escalate. A worker is just "role X instantiated on backend Y". This is exactly what makes QM model-agnostic workable — you swap the backend without re-architecting the role, and you can A/B two models on the same task type safely because the permissions and eval gates do not change.

flowchart TD
    A[Slack / Web / Email intake] --> B[task_intake_router]
    B --> C{role router}
    C -->|engineering| D[eng agent]
    C -->|finance| E[fin agent]
    C -->|legal| F[legal agent]
    C -->|ops| G[ops agent]
    D --> H{approval board}
    E --> H
    F --> H
    G --> H
    H -->|cross-role or risky| I[human approve]
    H -->|auto-ok| J[eval gate]
    I --> J
    J -->|pass| K[executor]
    J -->|fail| L[rework queue]
    L --> C
    K --> M[observability + audit]
    M --> N{self-improve?}
    N -->|yes| A
    N -->|no| O[Done]

Notice the self-improvement loop in the bottom corner: the harness does not end at completion. If the observability layer detects a recurring failure signature, the audit node files a new task back into the intake queue — "refactor the payment validation helper" is as legitimate a task as "fix the staging deploy", and it keeps the system improving itself the way YC's own QM develops QM.

Configuration: .env

# Backends (model-agnostic: pick per role, not globally)
BACKEND_PI_API=...        # Pi
BACKEND_OPENCODE_API=...  # OpenCode
BACKEND_CODEX_API=...     # Codex
BACKEND_CLAUDE_API=...    # Claude Code
DEFAULT_BACKEND=claude-code

# Intake channels
SLACK_BOT_TOKEN=xoxb-...
SLACK_SIGNING_SECRET=...
WEB_HOOK_URL=https://harness.example/tasks
EMAIL_IMAP_HOST=imap.example.com

# Role routing
ROLE_ENGINEERING_MODELS=open-code,claude-code
ROLE_FINANCE_MODELS=pi,claude-code
ROLE_LEGAL_MODELS=pi
ROLE_OPS_MODELS=claude-code

# Governance
AUTO_OK_THRESHOLD=0.9          # eval score above which no human needed
CROSS_ROLE_ALWAYS_APPROVE=true
APPROVAL_BOARD_CHANNEL=#approvals

# Observability
OTEL_EXPORTER=otlp://otel-collector:4317
AUDIT_SINK=postgresql://audit:audit@localhost:5432/audit
LOG_LEVEL=INFO

The role→model mapping is the crux of model-agnosticism done right. You do not pick one model for the company; you pick a default and a permitted set per role, and the role router picks from the set. Legal gets Pi (best at long-context contractual reasoning), engineering gets OpenCode and Claude Code, and ops defaults to whatever is cheapest that still passes the eval gate.

Schemas: schemas.py

from __future__ import annotations
from datetime import datetime, timezone
from enum import Enum
from typing import Any

from pydantic import BaseModel, Field

class Role(str, Enum):
    ENGINEERING = "engineering"
    FINANCE = "finance"
    LEGAL = "legal"
    OPS = "ops"

class Backend(str, Enum):
    PI = "pi"
    OPENCODE = "open-code"
    CODEX = "codex"
    CLAUDE_CODE = "claude-code"

class TaskSource(str, Enum):
    SLACK = "slack"
    WEB = "web"
    EMAIL = "email"

class TaskPriority(str, Enum):
    P0 = "p0"
    P1 = "p1"
    P2 = "p2"

class Task(BaseModel):
    task_id: str
    source: TaskSource
    channel: str                    # slack channel / web form id / mailbox
    author: str
    title: str
    body: str
    priority: TaskPriority = TaskPriority.P2
    labels: list[str] = Field(default_factory=list)
    routed_role: Role | None = None
    assigned_backend: Backend | None = None
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

class RoleManifest(BaseModel):
    role: Role
    skills: list[str] = Field(default_factory=list)
    allowed_backends: list[Backend]
    permissions: dict[str, list[str]] = Field(default_factory=dict)  # resource -> verbs
    must_escalate: list[str] = Field(default_factory=list)
    auto_ok_threshold: float = 0.9

class ApprovalRequest(BaseModel):
    approval_id: str
    task_id: str
    reason: str                     # cross_role | high_risk | low_confidence | irreversible
    role: Role
    risk_level: str                 # low | medium | high
    requested_by: str               # agent id
    status: str = "pending"         # pending | approved | rejected
    decided_by: str | None = None
    decided_at: datetime | None = None

class EvalResult(BaseModel):
    task_id: str
    role: Role
    backend: Backend
    score: float = 0.0              # 0..1 eval-gate score
    criteria: dict[str, float] = Field(default_factory=dict)
    passed: bool = False

class AgentRun(BaseModel):
    run_id: str
    task_id: str
    agent_id: str
    role: Role
    backend: Backend
    model_name: str
    tokens_in: int = 0
    tokens_out: int = 0
    cost_inr: float = 0.0
    started_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    finished_at: datetime | None = None

class AuditEntry(BaseModel):
    entry_id: str
    run_id: str
    task_id: str
    actor: str
    action: str                     # intaked | routed | approved | eval | executed | self_filed
    detail: dict[str, Any] = Field(default_factory=dict)
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

RoleManifest is where governance lives: permissions maps a resource to the verbs a role may perform, must_escalate lists actions that can never be auto-ok'd, and auto_ok_threshold sets how much confidence an eval must show before a human approval is skipped. A finance agent with permissions={"payments": ["read"]} cannot pay anything — it can only review.

External I/O with retry: tools.py

import asyncio
import random
from datetime import datetime, timezone

import httpx

class BackendDown(RuntimeError):
    pass

class NoBackendAvailable(BackendDown):
    pass

async def with_retry(coro_factory, *, attempts=4, base=0.3, max_backoff=5.0):
    for attempt in range(1, attempts + 1):
        try:
            return await coro_factory()
        except (httpx.TimeoutException, httpx.HTTPStatusError) as exc:
            if attempt == attempts:
                raise BackendDown(str(exc)) from exc
            delay = min(max_backoff, base * (2 ** (attempt - 1))) * (1 + random.uniform(0, 0.3))
            await asyncio.sleep(delay)

BACKENDS = {
    Backend.PI: {"url": "https://api.pi.example", "api_key_env": "BACKEND_PI_API"},
    Backend.OPENCODE: {"url": "https://api.opencode.example", "api_key_env": "BACKEND_OPENCODE_API"},
    Backend.CODEX: {"url": "https://api.codex.example", "api_key_env": "BACKEND_CODEX_API"},
    Backend.CLAUDE_CODE: {"url": "https://api.claude.example", "api_key_env": "BACKEND_CLAUDE_API"},
}

def backend_client(backend: Backend) -> httpx.AsyncClient:
    cfg = BACKENDS[backend]
    return httpx.AsyncClient(base_url=cfg["url"],
                             headers={"Authorization": f"Bearer {os.getenv(cfg['api_key_env'])}"})

async def call_model(backend: Backend, messages: list[dict], role: Role) -> dict:
    """Model-agnostic adapter: same request shape, backend-specific auth/endpoint."""
    async def call():
        async with backend_client(backend) as client:
            r = await client.post("/v1/chat/completions",
                                  json={"model": os.getenv(f"MODEL_{backend.value.upper()}"),
                                        "messages": messages,
                                        "role": role.value,
                                        "temperature": 0.2})
            r.raise_for_status()
            return {"text": r.json()["choices"][0]["message"]["content"],
                    "usage": r.json()["usage"]}
    return await with_retry(call)

async def send_slack(channel: str, text: str) -> None:
    async def call():
        async with httpx.AsyncClient(
                headers={"Authorization": f"Bearer {os.getenv('SLACK_BOT_TOKEN')}"}) as client:
            r = await client.post("https://slack.com/api/chat.postMessage",
                                  json={"channel": channel, "text": text})
            r.raise_for_status()
    await with_retry(call, attempts=3)

async def approve_via_board(request: ApprovalRequest) -> ApprovalRequest:
    """Post to #approvals, poll for a human verdict, timeout after 8h."""
    await send_slack(os.getenv("APPROVAL_BOARD_CHANNEL"),
                     f"Approval {request.approval_id}: {request.reason} "
                     f"for {request.role.value} task {request.task_id}")
    deadline = datetime.now(timezone.utc).timestamp() + 8 * 3600
    while datetime.now(timezone.utc).timestamp() < deadline:
        status = await check_approval_status(request.approval_id)
        if status in ("approved", "rejected"):
            request.status, request.decided_at = status, datetime.now(timezone.utc)
            return request
        await asyncio.sleep(30)
    request.status = "rejected"   # timeout = silent reject
    return request

async def run_eval_gate(task: Task, role: Role, backend: Backend,
                        agent_output: str) -> EvalResult:
    """Deterministic gates first (schema/perms), model-grade last."""
    criteria = {}
    criteria["schema_valid"] = 1.0 if agent_output else 0.0
    criteria["perm_respect"] = await check_permission_compliance(task, role, agent_output)
    criteria["factual_grounding"] = await grade_grounding(task, role, backend, agent_output)
    score = sum(criteria.values()) / len(criteria)
    return EvalResult(task_id=task.task_id, role=role, backend=backend,
                      score=score, criteria=criteria,
                      passed=score >= role_manifest(role).auto_ok_threshold)

async def append_audit(entry: AuditEntry) -> None:
    async def call():
        async with httpx.AsyncClient() as client:
            r = await client.post(os.getenv("AUDIT_SINK"), json=entry.model_dump())
            r.raise_for_status()
    await with_retry(call, attempts=3)

The adapter (call_model) is the whole game of model-agnosticism in one function: request shape is normalized, auth is backend-specific, and the routing graph never knows or cares which vendor answered. If you want to follow the broader ecosystem of such adapters, our MCP directory tracks the tooling layer underneath them.

The graph: graph.py

from typing import TypedDict

from langgraph.graph import END, StateGraph

from schemas import ApprovalRequest, AuditEntry, EvalResult, Task
from tools import (append_audit, approve_via_board, call_model, run_eval_gate,
                   send_slack)

class HarnessState(TypedDict):
    task: Task
    route: str
    role_manifest: dict
    agent_output: str | None
    approval: ApprovalRequest | None
    eval: EvalResult | None
    audit: list[AuditEntry]

async def intake_router(state: HarnessState):
    """Normalize Slack/web/email into a Task; classify source + priority."""
    task = state["task"]
    task.priority = classify_priority(task.title, task.body)
    await append_audit(AuditEntry(entry_id=uuid4().hex, run_id=state["run_id"],
                                  task_id=task.task_id, actor="intake",
                                  action="intaked", detail={"priority": task.priority.value}))
    return {"task": task}

def role_router(state: HarnessState) -> str:
    """Deterministic classifier → Role, then pick a backend from the role's set."""
    role = classify_role(state["task"])          # keyword + embedding classifier
    backends = ROLE_MODELS[role.value]
    state["task"].routed_role = role
    state["task"].assigned_backend = select_backend(backends, state["task"])
    state["route"] = role.value
    return role.value

async def run_agent(state: HarnessState):
    role, backend = state["task"].routed_role, state["task"].assigned_backend
    prompt = build_role_prompt(state["task"], role_manifest(role))
    resp = await call_model(backend, [{"role": "user", "content": prompt}], role)
    state["agent_output"] = resp["text"]
    return {"agent_output": resp["text"]}

def approval_router(state: HarnessState):
    """Auto-ok vs board: cross-role, high-risk, or low-confidence goes to humans."""
    req = build_approval_request(state["task"], state["role_manifest"], state["eval"])
    if req is None:
        return "eval"      # no approval needed
    return "approval"

async def approval_board(state: HarnessState):
    state["approval"] = await approve_via_board(state["approval"])
    return {"approval": state["approval"]}

def route_after_approval(state: HarnessState) -> str:
    if state["approval"].status == "approved":
        return "eval"
    return "done_rejected"

async def eval_gate(state: HarnessState):
    state["eval"] = await run_eval_gate(state["task"], state["task"].routed_role,
                                        state["task"].assigned_backend, state["agent_output"])
    return {"eval": state["eval"]}

def route_after_eval(state: HarnessState) -> str:
    if state["eval"].passed:
        return "execute"
    return "rework"

async def executor(state: HarnessState):
    """Apply approved+verified output: merge PR, post invoice, file form, deploy."""
    result = await apply_output(state["task"], state["agent_output"])
    await append_audit(AuditEntry(entry_id=uuid4().hex, run_id=state["run_id"],
                                  task_id=state["task"].task_id, actor="executor",
                                  action="executed", detail={"result": result}))
    return {}

def rework_queue(state: HarnessState):
    """Failures file a tighter-scoped task back to intake (self-improvement)."""
    new_task = scope_down_task(state["task"], state["eval"])
    await file_to_intake(new_task)          # re-enters intake_router
    await append_audit(AuditEntry(entry_id=uuid4().hex, run_id=state["run_id"],
                                  task_id=state["task"].task_id, actor="eval",
                                  action="self_filed",
                                  detail={"score": state["eval"].score}))
    return {}

async def observability(state: HarnessState):
    """Unified trace + audit close-out; may file improvement tasks."""
    await push_trace(state)
    await append_audit(AuditEntry(entry_id=uuid4().hex, run_id=state["run_id"],
                                  task_id=state["task"].task_id, actor="otel",
                                  action="observed", detail={"eval": state["eval"].model_dump()}))
    return {}

builder = StateGraph(HarnessState)
builder.add_node("intake", intake_router)
builder.add_node("eng", run_agent)
builder.add_node("fin", run_agent)
builder.add_node("legal", run_agent)
builder.add_node("ops", run_agent)
builder.add_node("approval", approval_board)
builder.add_node("eval", eval_gate)
builder.add_node("execute", executor)
builder.add_node("rework", rework_queue)
builder.add_node("observe", observability)

builder.set_entry_point("intake")
builder.add_conditional_edges("intake", role_router,
                              {"engineering": "eng", "finance": "fin",
                               "legal": "legal", "ops": "ops"})
for r in ("eng", "fin", "legal", "ops"):
    builder.add_edge(r, "approval_router")   # 2-node micro-router
builder.add_edge("approval", "eval")
builder.add_conditional_edges("eval", route_after_eval,
                              {"execute": "execute", "rework": "rework"})
builder.add_edge("execute", "observe")
builder.add_edge("observe", END)
builder.add_edge("rework", END)

graph = builder.compile()

One deliberate simplification to call out: run_agent is registered under four names (eng, fin, legal, ops) but is the same function — role differences live in the prompt built from role_manifest, not in the node. That keeps the permission matrix in one place, which is what makes a company-wide harness auditable instead of a tangle of per-department glue.

Entrypoint: main.py

import asyncio
import os

from dotenv import load_dotenv
from graph import graph
from schemas import Task

load_dotenv()

async def run_tasks(tasks: list[Task]) -> None:
    for t in tasks:
        initial = {"task": t, "route": None, "role_manifest": ROLE_MANIFESTS,
                   "agent_output": None, "approval": None, "eval": None, "audit": []}
        print(f"--- {t.task_id}: {t.title} ({t.source.value}) ---")
        final = await graph.ainvoke(initial)
        tsk = final["task"]
        print(f"role={tsk.routed_role.value} backend={tsk.assigned_backend.value} "
              f"eval={final['eval'].score:.2f} status={final['eval'].passed}")

async def main():
    await run_tasks([
        Task(task_id="T-5001", source="slack", channel="#eng", author="priya",
             title="staging deploy failed on payment-mock", body="see trace ...",
             labels=["deploy", "payments"]),
        Task(task_id="T-5002", source="web", channel="finance-form", author="rahul",
             title="reconcile vendor invoices Q2", body="uploaded 47 PDFs ..."),
        Task(task_id="T-5003", source="email", channel="legal@inbox", author="legal-team",
             title="draft NDA rider for EV partnership", body="redline attached ..."),
    ])

if __name__ == "__main__":
    asyncio.run(main())

Three tasks, three roles, three backends — one harness. The intake router saw a Slack message, a web form, and an email and turned all three into the same Task shape; the role router sent them to engineering, finance, and legal; and each ran on a backend chosen from that role's permitted set. That is the QM thesis: the orchestration loop is the product, and the models underneath are interchangeable workers.

Retry Rules & Error Handling

The model-backend row is the most company-specific: when OpenCode is slow, the fallback is not "wait longer" — it is Codex. Because the adapter normalizes everything, backend failover is a config change, not a refactor. That is the practical payoff of model-agnosticism.

Cost & Decision Matrix

Run the economics once and the pattern is obvious: routing a task through the classifier costs about the same as a single model token, while a wrong role routing costs a full wasted run plus a board review. The harness should spend aggressively on routing quality and stingily on everything downstream — which is exactly the opposite of most teams' instinct to spend on the biggest model and skip the routing layer.

What a company harness means for 2026 teams

YC open-sourcing QM under MIT is a strong signal that "multiplayer" agent infrastructure is becoming a commodity you run yourself rather than buy as a closed platform. The dividing line between the products that will survive and the ones that won't is no longer model quality — every vendor has a good model — it is governance: roles, routing, permissions, approvals, and audit. In Indian enterprise reality that means the harness is also your compliance story: an agent that touches invoices and NDAs must be explainable, permission-scoped, and fully logged, and the eval gates give your auditors a recorded confidence score for every automated decision. For more production patterns like this, browse the full library in AI Workflows, track the agent-tooling wave in our MCP directory, and don't miss the weekly roundups in Latest AI News.

Ship checklist

• Roles before agents. Define what Finance may not do before you build what it can do.
• Model-agnostic adapter from day one. The swap cost between backends must be a config line.
• Approval board for cross-role and high-risk. Auto-ok only above a measured eval threshold.
• Eval gates on everything. A recorded score is your audit story and your rework signal.
• Self-improvement loop with a rework cap. Agents fix their own tooling — but max two re-files, then a human.

QM shows the destination: a company where accounting, legal, events, and engineering run through one multiplayer harness, each department on the model that suits it, every decision gated and logged. The harness is not a wrapper around an LLM — it is the operating system of the company's work. Build the roles, the routing, and the audit first, and the models become the most replaceable part of the whole system.