Agent frameworks

LangChain and LangGraph

vincs[langgraph] brings the relevant part of your Vinc graph into LangGraph nodes and LangChain agents. It reads only; writing back happens in a step you place after a person approves.

Install

pip install "vincs[langgraph]"

Requires Python 3.11 or later. This extra installs LangGraph, langchain-core and LangChain 1.0 or later, including the create_agent middleware. Set VINC_API_KEY to a read-only key (vinc_ro_); only record_episode and write_package need a write key, and with a read-only key they raise before anything is sent.

This page describes vincs 0.1.5, which carries vinc-langgraph 0.1.9. The separate vinc-langgraph package has the same names under vinc_langgraph.

Role mode or context mode

VincContextMiddleware runs in role mode when you pass prefetch_role, and in context mode when you do not.

The per-run read and the tool rule come with vincs 0.1.4. With vincs 0.1.3 the middleware reads the role before every model call and does not apply disallowed_tools. In a graph node, prefetch raises like role mode and get_vinc_context returns None like context mode.

In a graph node

Call get_vinc_context at the start of the node that calls the model, and add the block to the system message with build_system_message.

from langchain.chat_models import init_chat_model
from vinc import VincClient
from vinc.integrations.langgraph import build_system_message, get_vinc_context

llm = init_chat_model("openai:gpt-5")
vinc = VincClient()                                    # VINC_API_KEY=vinc_ro_...

def review(state):
    block = get_vinc_context(vinc, state["messages"])   # newest human message
    system = build_system_message(block, "You review design changes.")
    return {"messages": [llm.invoke([system, *state["messages"]])]}

• get_vinc_context(client, messages) takes the newest human message from a list of messages, or a plain string. It returns None when nothing matched, or when Vinc is unreachable or the daily limit is used up, so the node goes on without it.
• It makes one call, or two when nothing matched or several nodes tied: it then widens with a search. Pass search_fallback=False to keep it to one.
• build_system_message(block, base) appends the block to base (a string or a SystemMessage) and returns base unchanged when there is no block.
• For async graphs, use aget_vinc_context with an AsyncVincClient.

In a create_agent agent

Add VincContextMiddleware and every model call gets the block in its system message.

from langchain.agents import create_agent
from vinc.integrations.langgraph import VincContextMiddleware

agent = create_agent("openai:gpt-5", middleware=[VincContextMiddleware()])
agent.invoke({"messages": [{"role": "user", "content": "Why are tokens OKLCH?"}]})

Tools the model can call

create_vinc_tools gives the model two read-only tools: vinc_brief (resolve a question to one node with its material) and vinc_search (search text and nodes, with a limit). Pass a sync client, an async client, or both.

from vinc import VincClient
from vinc.integrations.langgraph import create_vinc_tools

tools = create_vinc_tools(VincClient())   # vinc_brief and vinc_search, read only
agent = create_agent("openai:gpt-5", tools=tools)

Record after a person approves

Put record_episode in the node after an interrupt(), and give only that node a write key. The graph below reads, drafts, waits for a person and records only what was approved.

context → review → approve (interrupt) → record

import os
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from vinc import VincClient
from vinc.integrations.langgraph import get_vinc_context, record_episode

reader = VincClient()                                   # vinc_ro_ key
writer = VincClient(os.environ["VINC_WRITE_KEY"])      # vinc_sk_ key, record step only

class State(TypedDict, total=False):
    question: str
    context: str | None
    draft: str
    approved: bool
    about: list[str]
    written: dict | None

def context(state):  return {"context": get_vinc_context(reader, state["question"])}
def review(state):   return {"draft": draft_review(state["question"], state["context"])}  # your model call
def approve(state):  return {"approved": bool(interrupt({"draft": state["draft"]}))}
def record(state):
    if not state["approved"]:
        return {"written": None}
    return {"written": record_episode(writer, state["draft"], about=state["about"])}

g = StateGraph(State)
for name, fn in [("context", context), ("review", review), ("approve", approve), ("record", record)]:
    g.add_node(name, fn)
g.add_edge(START, "context"); g.add_edge("context", "review")
g.add_edge("review", "approve"); g.add_edge("approve", "record"); g.add_edge("record", END)
app = g.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "review-1"}}
paused = app.invoke({"question": "Does the new button meet contrast?", "about": []}, cfg)
app.invoke(Command(resume=True), cfg)                  # a person approved

Retry the record step

From vincs 0.1.4, record_episode, arecord_episode, write_package and awrite_package take create_only and operation_id and pass them to the client unchanged.

• Before the interrupt(), store a fresh lowercase UUID next to the episode id in the LangGraph state, and pass it as operation_id on every attempt of that write. An identical retry returns the stored result without applying again.
• With an operation_id, the client also retries the write itself while the hub is briefly down. After an uncertain response (VincUnavailable), read writer.operation_status(operation_id) before deciding. accepted and unknown do not prove the write did not happen.
• With create_only=True, an episode id that already exists is refused. Retry such a write with the same operation_id, or the retry is refused because the first attempt created the episode.
• Delete and merge are not wrapped here. Call the client's delete, delete_edge, merge_nodes or doc_delete with dry_run=True first, then apply with the preview_digest it returns. See the Python SDK.

Run the agent as a Vinc role

Pass prefetch_role to VincContextMiddleware. On the run's first model call it reads that role's whole contract, the full text of the documents the role says to read first, and Vinc's material on the task, with node ids the model can cite, and every model call of the run gets that block. The model gets the evidence without having to decide to look it up.

from langchain.agents import create_agent
from vinc import VincClient
from vinc.integrations.langgraph import VincContextMiddleware, create_vinc_tools

reader = VincClient()                                  # vinc_ro_ key
agent = create_agent(
    "openai:gpt-5",
    tools=create_vinc_tools(reader),                   # optional, for further lookups
    middleware=[VincContextMiddleware(
        client=reader, prefetch_role="concept:agent-reviewer-judgement",
        prefetch_max_chars=120000,
    )],
)

Prefetch reads the complete role contract and each required document by its declared document id once per run; a new human message starts a new run and reads again. Each document must have complete text and a verified content hash; missing, partial or unverified documents raise prefetch_spec_unresolved before optional task lookup. Authentication and service errors propagate, so the run stops before the model is called. The combined prefetch_max_chars budget defaults to 24,000 characters. Large roles need a larger budget, such as the 120,000-character budget in this example. prefetch_budget_exceeded stops the read instead of truncating the contract or required documents. Your code picks the role and its budget. Helpers use max_chars for this same budget.

To keep the agent on the contract you reviewed, pin the role. Read it once, keep its hash, and pass it as prefetch_pin. When someone has changed the role (its props, title, topic or relations), the run's first model call raises VincRoleChanged with the current rev and hash in details, and the model is not called. Moving the pin to the new hash is your decision. The pin covers the role contract, not the text of the documents it lists. Graph node helpers take the same value as pin. It comes with vincs 0.1.5.

pin = reader.role("concept:agent-reviewer-judgement")["hash"]    # after you reviewed it
middleware = VincContextMiddleware(
    client=reader, prefetch_role="concept:agent-reviewer-judgement",
    prefetch_max_chars=120000, prefetch_pin=pin,
)

Check the final answer against the role

OutputGate checks the final answer outside the model, against the role's gates, never_checks and person_only. It answers pass, block or interrupt, and the middleware raises OutputGateStopped on the last two. It calls no model and writes nothing.

from vinc import OutputGate
from vinc.integrations.langgraph import create_output_gate_middleware

role = reader.role("concept:agent-reviewer")           # read fresh for this run
gate = OutputGate(role["props"])
agent = create_agent("openai:gpt-5", middleware=[create_output_gate_middleware(gate)])
agent.invoke({"messages": [{"role": "user", "content": "Review the proposed change."}]})

• Use invoke or ainvoke. Do not show unchecked streamed tokens to a user.
• In a StateGraph, put evaluate_output_node(state, gate) after the model node and route on its verdict, or call require_output_pass to raise instead.
• A claim that something ran, such as tests passing, needs evidence your host saw in this run, passed as evidence. The model's own word never counts.
• A role whose contract is prose only, or names a check the gate does not know, blocks. The gate stops output; it does not grant or take away tools. Role mode keeps tools named in disallowed_tools out of the run; leave any other tool the role must not use out of the agent.

Ask a person before a person-only step

Have the model node store a vinc.RoleOutput in role_output, add the node from create_person_only_node, and compile with a checkpointer. When the answer asks for a person-only action, the node pauses with a native interrupt. Resume it with the person's decision; the model is not called again.

from langgraph.types import Command
from vinc.integrations.langgraph import create_person_only_node

graph.add_node("approval", create_person_only_node(gate, role_id=role["id"], run_id="review-1"))
graph.add_edge("model", "approval")

paused = app.invoke(inputs, cfg)
pending = paused["__interrupt__"][0].value
app.invoke(Command(resume={"request_id": pending["request_id"], "approved": True}), cfg)

Declare vinc_approval in the state schema and deliver only that checked result. A rejected answer has nothing to deliver, and approval never overrides a block. Approval releases this answer only; it does not run or permit any tool.

Hand off to another agent

Passing only citation ids loses the first agent's conclusion. Instead, the first agent writes one package Record with the question, the conclusion and each claim's evidence, and the next agent reads it by id.

from vinc.integrations.langgraph import get_vinc_package_context, write_package

handoff = write_package(                               # first agent, vinc_sk_ key
    writer, "When does the dashboard become stale?", "Keep the 36 hour threshold.",
    claims=[{"claim": "Stale after 36 hours.", "evidence_ids": ["decision:threshold"]}],
    produced_by="agent:analyst", run_id="analysis-1", space="personal",
)
block = get_vinc_package_context(reader, handoff["package_id"], space=handoff["space"])

Name the space; the evidence must already exist there. The next agent rereads the package and checks every evidence id with its own key each time, so deleted or unreadable evidence is marked as not citable. For a create_agent agent bound to one handoff, add PackageProvider(package_id, space=..., client=reader) as middleware.

Structured review, one retry

run_review(model, messages) (async arun_review) asks the model for a ReviewOutput: a verdict of ACCEPT or RETURN, findings with citations, and person-only requests. A malformed answer gets one retry. A second failure raises UnparsedReview instead of passing as an empty review; hand that to a person.

from vinc.integrations.langgraph import UnparsedReview, run_review

try:
    review = run_review(model, "Review this draft against its cited rules")
except UnparsedReview:
    raise                                              # stop here and hand it to a person

Every helper

Import them from vinc.integrations.langgraph. A helper that reads or writes takes the client first; one whose name starts with a takes an AsyncVincClient and is awaited.

Over MCP instead

If you would rather not add a package, langchain-mcp-adapters can reach the Vinc MCP server. Keep the tool list to reads.

import os
from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({"vinc": {
    "transport": "http", "url": "https://mcp.vincs.io/mcp",
    "headers": {"Authorization": f"Bearer {os.environ['VINC_API_KEY']}"},
}})
tools = [t for t in await client.get_tools()
         if t.name in {"vinc_brief", "vinc_search"}]