langchain-agentkit

Composable extension framework for LangGraph agents.

Table of Contents

• Installation
• Quick Start
  • The Agent class
  • Dynamic properties
  • AgentKit for managed or manual graph wiring
• Extensions
  • Recommended ordering
  • SkillsExtension
  • AgentsExtension
  • TasksExtension
  • FilesystemExtension
  • WebSearchExtension
  • HistoryExtension
  • MemoryExtension
  • HITLExtension
  • TeamExtension
  • RubricExtension
• FilesystemExtension Backends
• Custom Extensions
  • Lifecycle hooks
  • Sibling-aware configuration via setup()
  • Declaring dependencies
• Contributing

Installation

pip install langchain-agentkit

Requires Python 3.12+.

Quick Start

The Agent class

Declare a class, get a complete ReAct agent with extension-composed tools and prompts:

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_agentkit import Agent, SkillsExtension, TasksExtension

class Researcher(Agent):
    model = ChatOpenAI(model="gpt-4o")
    extensions = [
        SkillsExtension(skills="skills/"),
        TasksExtension(),
    ]
    prompt = "You are a research assistant."

    async def handler(state, *, llm, tools, prompt):
        messages = [SystemMessage(content=prompt)] + state["messages"]
        return {"messages": [await llm.bind_tools(tools).ainvoke(messages)]}

app = Researcher().compile()
result = app.invoke({"messages": [HumanMessage("Size the B2B SaaS market")]})

The model attribute accepts a BaseChatModel instance (used as-is) or a string resolved via model_resolver:

class FastAgent(Agent):
    model = "gpt-4o-mini"
    model_resolver = staticmethod(lambda name: ChatOpenAI(model=name))
    extensions = [AgentsExtension(agents=[Researcher, Coder])]
    ...

The state schema is composed automatically from extensions — TasksExtension adds a tasks key, SkillsExtension adds nothing. No need to define state manually.

Dynamic properties

Each property (model, prompt, extensions, tools, model_resolver) can be a static attribute, a sync method, or an async method — compile() resolves them uniformly. Pass any per-request configuration via __init__(**kwargs):

from langchain_agentkit import Agent, FilesystemExtension

class Researcher(Agent):
    model = ChatOpenAI(model="gpt-4o")

    async def prompt(self):
        return await self.backend.read(".agentkit/AGENTS.md")

    def extensions(self):
        return [FilesystemExtension(backend=self.backend)]

    async def handler(state, *, llm, tools, prompt, runtime):
        messages = [SystemMessage(content=prompt)] + state["messages"]
        return {"messages": [await llm.bind_tools(tools).ainvoke(messages)]}

app = Researcher(backend=my_backend).compile()

Use graph() instead of compile() when you need the uncompiled StateGraph for composition — e.g. passing to AgentsExtension(agents=[...]) or embedding as a subgraph.

A compiled agent also drops straight into an external supervisor/router graph as a node: it shares the parent's messages channel, and a tool you pass it can return Command(goto=..., graph=Command.PARENT) to hand control back to the supervisor — no framework change required. See examples/supervisor.py for a complete example.

AgentKit for managed or manual graph wiring

AgentKit accepts extensions, user tools, model, and prompt — then compile(handler) builds a complete ReAct graph with hooks wired:

from langchain_openai import ChatOpenAI
from langchain_agentkit import AgentKit, SkillsExtension, TasksExtension

kit = AgentKit(
    extensions=[SkillsExtension(skills="skills/"), TasksExtension()],
    tools=[web_search],
    model=ChatOpenAI(model="gpt-4o"),
    prompt="You are a research assistant.",
)

async def handler(state, *, llm, tools, prompt, runtime):
    from langchain_core.messages import SystemMessage
    messages = [SystemMessage(content=prompt)] + state["messages"]
    return {"messages": [await llm.bind_tools(tools).ainvoke(messages)]}

graph = kit.compile(handler)          # uncompiled StateGraph
app = graph.compile()                 # compiled, ready to invoke

For full manual control over graph topology (custom routing, multi-node graphs, shared ToolNode), access kit.tools, kit.compose(state), kit.model, kit.state_schema, and kit.hooks directly:

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode

from langchain_agentkit import AgentKit, SkillsExtension, TasksExtension

kit = AgentKit(extensions=[
    SkillsExtension(skills="skills/"),
    TasksExtension(),
])

llm = ChatOpenAI(model="gpt-4o")
all_tools = kit.tools
bound_llm = llm.bind_tools(all_tools)

def agent_node(state):
    prompt = kit.compose(state).joined
    messages = [SystemMessage(content=prompt)] + state["messages"]
    return {"messages": [bound_llm.invoke(messages)]}

def should_continue(state):
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

# State schema composed automatically from extensions
graph = StateGraph(kit.state_schema)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(all_tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile()
result = app.invoke({"messages": [HumanMessage("Size the B2B SaaS market")]})

Extensions

Each extension provides tools, a prompt section, and optional state requirements. Compose them in any combination:

extensions = [
    SkillsExtension(skills="skills/"),
    TasksExtension(),
    FilesystemExtension(),
    WebSearchExtension(),
    HistoryExtension(strategy=CountStrategy(max_messages=50)),
    HITLExtension(interrupt_on={"send_email": True}),
    AgentsExtension(agents=[researcher, coder]),
    TeamExtension(agents=[researcher, coder]),
]

Recommended ordering

Declaration order in extensions=[...] determines prompt-section order in the composed system prompt. For best results, declare extensions in the following order:

extensions = [
    # --- Prompt-contributing layer (order shapes system prompt) ---
    CoreBehaviorExtension(),                        # universal behavior
    MemoryExtension(),                              # persistent user/project memory
    EnvExtension(),                                 # auto-detected runtime env block
    FilesystemExtension(root="..."),                # workspace root + file tools
    SkillsExtension(skills="skills/"),              # progressive-disclosure skill catalog
    TasksExtension(),                               # task tracking
    WebSearchExtension(),                           # external research tool
    TeamExtension(agents=[...]),                    # peer-to-peer coordination
    AgentsExtension(agents=[...]),                  # delegate-to-specialist tool

    # --- Hook-only layer (order shapes wrap onion, not prompt) ---
    ResilienceExtension(),                          # outermost wrap_tool — catches unhandled tool errors
    HITLExtension(interrupt_on={"send_email": True}),
    HistoryExtension(strategy=CompactionStrategy()),
    MessagePersistenceExtension(persist=write_to_db),
]

The stack has two layers. The prompt-contributing layer produces the system prompt in declaration order — placement here is visible to the model. The hook-only layer contributes no prompt text; declaration order here only determines wrap-hook nesting and run-lifecycle call order.

Rationale for each slot:

Notes:

• Dependencies are auto-resolved. Missing dependencies are prepended automatically — e.g. omitting TasksExtension while using TeamExtension still produces a valid stack.
• preset="full" seeds the first two slots. Passing preset="full" to AgentKit(...) prepends CoreBehaviorExtension and TasksExtension ahead of any user-declared extensions. Passing your own instance of a seeded type replaces the seeded one in its slot.
• Every extension listed is opt-in. Include only what your agent needs. Hook-layer extensions (HITL, History, MessagePersistence) pay off in different scenarios: use History for long sessions, HITL when specific tools need approval, and MessagePersistence when turns must be mirrored to an external store.
• Wrap-hook ordering is an onion. First declaration = outermost layer. ResilienceExtension must stay ahead of any other wrap_tool extension so it can catch exceptions raised inside them. It also repairs tool calls left without a result, including calls whose arguments were malformed or truncated.

SkillsExtension

Loads skills and provides progressive disclosure — the agent sees skill names and descriptions, then loads full content on demand via the Skill tool.

Two input modes:

from langchain_agentkit import SkillsExtension, SkillConfig

# Programmatic — pass SkillConfig objects directly
ext = SkillsExtension(skills=[
    SkillConfig(name="market-sizing", description="Calculate TAM/SAM/SOM", prompt="..."),
])

# Directory discovery — scan a directory for SKILL.md files
ext = SkillsExtension(skills="skills/")

# With a custom backend (e.g. Daytona sandbox or AgentFS database)
ext = SkillsExtension(skills="/skills", backend=my_backend)

Always provides exactly one tool: Skill. Filesystem tools (Read, Write, etc.) come from FilesystemExtension.

Tools:

Pinned skills — an application can load skills deterministically instead of waiting for the model to call Skill. Name them in the pinned_skills state key; before the next model call each skill's content is added to the conversation once and the key is cleared:

await graph.ainvoke({"messages": [HumanMessage("Size this market")], "pinned_skills": ["market-sizing"]})

Unknown names are logged and skipped. Descriptions longer than the roster cap log a warning at discovery.

Skill directories follow the AgentSkills.io format:

skills/
└── market-sizing/
    ├── SKILL.md          # YAML frontmatter (name, description) + prompt body
    └── calculator.py     # Reference files accessible via Read tool

AgentsExtension

Delegate tasks to specialist subagents at runtime. Accepts compiled StateGraphs, AgentConfig definitions, or discovers agents from a directory of markdown files.

from langchain_agentkit import Agent, AgentsExtension, AgentConfig

class Researcher(Agent):
    model = ChatOpenAI(model="gpt-4o-mini")
    description = "Research specialist for information gathering"
    tools = [web_search]
    prompt = "You are a research specialist."
    async def handler(state, *, llm, tools, prompt): ...

researcher = Researcher().compile()  # StateGraph

# Programmatic — mix compiled graphs and AgentConfig definitions
ext = AgentsExtension(agents=[
    researcher,                                          # compiled StateGraph
    AgentConfig(name="coder", description="Code expert", prompt="You code."),
])

# Directory discovery — scan for .md files with frontmatter
ext = AgentsExtension(agents="agents/")

# With a custom backend
ext = AgentsExtension(agents="/agents", backend=my_backend)

AgentConfig supports the same frontmatter fields as file-based agents:

AgentConfig(
    name="researcher",
    description="Research specialist",
    prompt="You are a research assistant.",
    model="gpt-4o-mini",            # resolved via model_resolver
    tools=["WebSearch", "Read"],     # filtered from parent's tools
    skills=["api-conventions"],      # preloaded into prompt at delegation time
    max_turns=10,                    # hard recursion ceiling; unset → inherit parent's budget
    mode="isolated",                 # "isolated" (default) or "fork"
)

File-based agent (agents/researcher.md):

---
name: researcher
description: Research specialist
model: gpt-4o-mini
tools: WebSearch, Read
skills: api-conventions, error-handling
maxTurns: 10
---
You are a research assistant.

The Agent tool uses shape-based discrimination — the LLM provides either {id: "<name>"} for a pre-defined agent or {prompt: "..."} for a dynamic one:

{"agent": {"id": "researcher"}, "message": "Find info on X"}
{"agent": {"prompt": "You are a legal expert..."}, "message": "Analyze this contract"}

Unknown argument keys are rejected with a tool error that names the key, so a task placed under an invented key never reaches the subagent as an empty request.

Context modes:

Resuming a subagent — each result carries an agent id. Passing it back continues that subagent's conversation with its full history:

{"agent": {"id": "researcher"}, "resume": "<agent id>", "message": "Now check the 2025 figures"}

Child conversations are saved through the parent's checkpointer, scoped to the parent thread, so one conversation cannot resume another's children. Without a checkpointer, results carry no id and resume returns an error.

Approvals inside subagents — an interrupt() raised in a subagent (for example by a HITLExtension on the subagent) surfaces at the parent graph. Resuming the parent with Command(resume=...) finishes the subagent.

Provenance — the task a subagent receives names the delegating agent, and the Agent tool result is the subagent's report, so neither is mistaken for a message from the user.

Key features:

• description — used in the prompt roster so the LLM knows what each specialist does
• tools="inherit" — subagent receives the parent's tools at delegation time
• ephemeral=True — enables dynamic (on-the-fly) reasoning agents
• skills preloading — full skill content injected into agent's prompt at startup
• model override — per-agent model selection via model_resolver
• delegation_timeout — max seconds per delegation (default 300s)
• max_turns inheritance — a subagent that sets no max_turns/maxTurns inherits the outer agent's turn budget: a parent TurnBudgetExtension (soft, graceful wrap-up) and/or the parent's hard ceiling (Agent/AgentKit max_turns). An explicit per-subagent value is a hard ceiling and overrides inheritance. A subagent that hits its hard ceiling returns a distinct "reached its turn limit" result rather than a generic error.

See examples/delegation.py for a complete example.

TasksExtension

Task tracking for complex multi-step objectives. The agent creates, tracks, and completes tasks with dependency ordering. Tasks are a passive todo list; they do not execute work.

ext = TasksExtension()
ext.tools  # [TaskCreate, TaskUpdate, TaskList, TaskGet]

Tools:

Tasks support blocked_by dependencies, owner assignment, and arbitrary metadata. Parallel TaskCreate calls are handled by a merge-by-ID reducer.

FilesystemExtension

File tools operating on the OS filesystem via OSBackend:

from langchain_agentkit import FilesystemExtension

# Current working directory
ext = FilesystemExtension()

# Scoped to a specific directory (with path traversal prevention)
ext = FilesystemExtension(root="./workspace")

Tools:

Glob and Grep stop at a result cap and tell the model when results are partial; search failures are reported instead of looking like "no matches". Writes and edits to the same file within one step run one after another. When the active model's profile says it cannot accept images or PDFs, those blocks are replaced with a short placeholder for that request only; saved state keeps the originals.

Alternative backends: FilesystemExtension accepts any object satisfying FilesystemProtocol. The framework ships four optional remote/sandboxed/unified-VFS backends alongside the default OSBackend — see the FilesystemExtension Backends section for the full catalog and runnable examples.

WebSearchExtension

Web search returning typed SearchResult records. Providers are queried in fallback order by default — the first to return results wins — so with a single provider there is no duplication. Set fan_out=True to query every provider concurrently and merge, dedupe, and rank the results. Ships with two built-in providers (no API key needed):

from langchain_agentkit import WebSearchExtension, DuckDuckGoSearchProvider

# Zero config (defaults to DuckDuckGo)
ext = WebSearchExtension()

# DuckDuckGo HTML mode — broader general-web coverage
ext = WebSearchExtension(providers=[DuckDuckGoSearchProvider(use_html=True)])

# Multiple providers, queried concurrently and merged
from langchain_agentkit import QwantSearchProvider

ext = WebSearchExtension(
    providers=[DuckDuckGoSearchProvider(), QwantSearchProvider()],
    fan_out=True,
)

A provider is any SearchProvider returning list[SearchResult]. To use a backend (Tavily, Exa, Brave, …), subclass and map its response:

from langchain_agentkit import SearchProvider, SearchResult

class TavilyProvider(SearchProvider):
    name = "Tavily"

    def __init__(self, client):
        self._client = client

    def search(self, query: str, *, max_results: int) -> list[SearchResult]:
        hits = self._client.search(query, max_results=max_results)["results"]
        return [
            SearchResult(url=h["url"], title=h["title"], snippet=h["content"], score=h.get("score"))
            for h in hits
        ]

Set fetch=True to add a companion WebFetch tool that reads one URL as markdown — the model searches, then reads a result. Fetch needs the web extra (pip install langchain-agentkit[web]) for HTML→markdown conversion, and is SSRF-guarded (scheme allowlist, private/loopback/metadata IP blocking with IPv4-mapped and 6to4 unwrapping, DNS pinning against rebinding, per-hop redirect re-validation). For human approval on each fetch, compose HITLExtension(interrupt_on=["WebFetch"]).

# WebSearch + WebFetch (needs the [web] extra)
ext = WebSearchExtension(fetch=True)

HistoryExtension

Manage conversation history to keep the LLM context window lean. Rewritten history replaces graph state via ReplaceMessages so the checkpointer stays compact.

from langchain_agentkit import (
    CompactionStrategy,
    CountStrategy,
    HandoffStrategy,
    HistoryExtension,
    TokenStrategy,
)

# Keep the last 50 messages
ext = HistoryExtension(strategy=CountStrategy(max_messages=50))

# Keep messages within a token budget
ext = HistoryExtension(strategy=TokenStrategy(max_tokens=4000))

# Summarize older history when the context window fills up
ext = HistoryExtension(strategy=CompactionStrategy(reserve_tokens=16_384))

# Replace older history with a brief written toward the current request
ext = HistoryExtension(strategy=HandoffStrategy(reserve_tokens=16_384))

# Custom strategy — any object with async transform(messages, *, runtime)
ext = HistoryExtension(strategy=MyStrategy())

All built-in strategies preserve a leading SystemMessage. Compaction and handoff derive their previous summary from the conversation itself, so one strategy instance is safe to share across threads and subagents.

Overflow recovery — when the model call raises ContextOverflowError, a strategy that implements compact() (compaction and handoff do) shrinks the history and the call is retried once, only if the new request is smaller. Otherwise the original error propagates.

MemoryExtension

Persistent memory across sessions. The extension frames the content; a strategy decides where it comes from.

from langchain_agentkit import CuratedMemory, MemoryExtension, RecalledMemory

# An operator-authored MEMORY.md, always in the system prompt (default)
ext = MemoryExtension()  # ~/.agents/memory/<project-key>/MEMORY.md
ext = MemoryExtension(strategy=CuratedMemory(path="./memory", project_discovery=False))

# A store the agent writes and searches itself (needs the [memory] extra)
ext = MemoryExtension(strategy=RecalledMemory(namespace=("memories", "{user_id}"), auto_recall=3))

Declare the extension twice to combine both: a curated document that is always present, alongside a store the agent queries on demand.

RecalledMemory is backed by the LangGraph store passed to compile(store=...):

• The agent records and searches memories through tools; auto_recall=K adds the top K hits for the latest request to each user turn.
• extract=True runs a background reflector that writes memories from the conversation after a lull.
• To review the agent's own writes, gate its memory tool with HITLExtension(interrupt_on={"manage_memory": True}).
• To review background extraction, set staging_namespace. Extracted memories land there, invisible to recall, until you call approve_staged(key) or reject_staged(key); list_staged() shows what is pending.
• consolidate(source=..., target=...) merges duplicates and resolves contradictions into a separate namespace, leaving the source untouched so the result can be reviewed before it is swapped in. Run it from your own scheduler.

pip install 'langchain-agentkit[memory]'

HITLExtension

Human-in-the-loop via a unified Question protocol. Two capabilities:

Tool approval — gate sensitive tools with human review:

hitl = HITLExtension(interrupt_on={
    "send_email": True,           # approve / reject
    "delete_file": {"options": ["approve", "reject"]},
})
# Tools not listed in interrupt_on execute normally without interruption.

AskUser tool — the AskUser tool lets the LLM ask structured questions. It is provided by default:

hitl = HITLExtension()  # provides [AskUser]

# Combine with approval gating (AskUser is still included):
hitl = HITLExtension(interrupt_on={"send_email": True})

# Drop AskUser entirely:
hitl = HITLExtension(interrupt_on={"send_email": True}, tools=[])

# Replace AskUser with a custom tool:
hitl = HITLExtension(tools=[my_custom_tool])

Both use the same interrupt payload (Question objects) and resume format. Approval requests raised inside a subagent surface at the parent graph and resume the same way. Requires a checkpointer. Answers are keyed by each question's position in the interrupt questions array (as a string), so resume with Command(resume={"answers": {"0": "Approve"}}). For tool approval, "Approve" runs the tool and "Reject" denies it; any other value is forwarded to the LLM.

TeamExtension

Coordinate a team of concurrent agents for complex, multi-step work that requires back-and-forth communication. The lead spawns teammates, assigns tasks, reacts to their results, and can forward information between team members.

from langchain_agentkit import Agent, TeamExtension, TasksExtension

class Lead(Agent):
    model = ChatOpenAI(model="gpt-4o")
    extensions = [TasksExtension(), TeamExtension(agents=[researcher, coder])]
    prompt = "You are a project lead. Coordinate your team."
    async def handler(state, *, llm, tools, prompt): ...

graph = Lead().compile()

How it works: Teammates run as asyncio.Tasks with their own checkpointers (conversation history persists across messages). A Router Node in the graph checks for teammate messages after each tool execution — when a teammate sends a result, the lead is automatically re-invoked with the message. Messages between the lead and teammates name their sender and are marked as not coming from the user.

Teammates run outside the lead's graph step, so an approval request raised inside a teammate does not reach the lead; it is reported back as an error message.

Tools:

When to use Teams vs Agent:

See examples/team.py for a complete example.

RubricExtension

Rubric-gated iteration: a separate (often cheaper) grader sub-agent reviews the agent's work against a caller-supplied rubric and drives revision until every criterion is met. Pass the rubric on invocation state; with no rubric the extension is a no-op, so it's safe to include unconditionally.

from langchain_anthropic import ChatAnthropic
from langchain_agentkit.extensions.rubric import RubricExtension, ReviewPolicy

ext = RubricExtension(
    model=ChatAnthropic(model="claude-haiku-4-5"),  # grader (separate from the agent's model)
    mode="auto",                                     # auto | user
    review=ReviewPolicy(min=1, max=5, stop=7),
)
# invoke with:  {"messages": [...], "rubric": "- criterion A\n- criterion B"}

Modes — what happens when the grader returns needs_revision:

review — when the grader runs (an int, a dict, or a ReviewPolicy):

review=1 grades every turn; review=5 expands to (min=1, max=5). A request_review tool is offered in user mode when a discretionary window exists (max > min).

The grader runs off-graph and uses tool-calling structured output, so it works across providers. Its _rubric_* bookkeeping is marked PrivateStateAttr — kept out of the invoke result; observe the verdict via get_state, the on_evaluation callback, or rubric_evaluation_* stream events. Terminal statuses: satisfied, failed, review_limit_reached, grader_error.

In auto mode the revision loop routes via jump_to, honored only when the kit exposes at least one tool; the rubric's review.stop should sit within the agent's max_turns ceiling (see Recommended ordering).

See examples/rubric.py for a complete example.

FilesystemExtension Backends

The FilesystemExtension above operates on top of a backend that implements FilesystemProtocol (file ops + bulk transfer) and optionally SandboxProtocol (adds execute and environment). Capability gating is structural: the Bash tool registers automatically when the backend implements SandboxProtocol. Different backends address different deployment shapes — host-local development, cloud sandboxes, persistent virtual filesystems, multi-tenant Linux hosts, or unified multi-service VFS.

# Pick any backend that satisfies the protocol you need
from langchain_agentkit.backends.daytona import DaytonaBackend     # cloud sandbox
# from langchain_agentkit.backends.agentfs import AgentFSBackend    # SQLite-backed VFS
# from langchain_agentkit.backends.bubblewrap import BubblewrapBackend  # Linux sandbox
# from langchain_agentkit.backends.mirage import MirageBackend      # multi-mount VFS

ext = FilesystemExtension(backend=...)

glob() and grep() return GlobResult / GrepResult with truncated and error_message, and every backend applies the same glob rules. Remote backends run searches with python3 inside the sandbox.

Each example under examples/<backend>/ ships a runnable end-to-end agent against that backend; see the per-example README for upstream links, prerequisites, and the recommended composition pattern.

Custom Extensions

Any subclass of Extension can contribute tools, a prompt section, state schema, lifecycle hooks, and graph modifications:

from langchain_agentkit import Extension

class MyExtension(Extension):
    @property
    def tools(self):
        return [my_tool]

    def prompt(self, state, runtime=None):
        return "You have access to my_tool."

    @property
    def state_schema(self):
        return None  # or a TypedDict mixin

Lifecycle hooks

Extensions intercept the run/model/tool lifecycle via three decorators:

wrap_run is intentionally not supported. before_run and after_run are implemented as langgraph nodes (not Runnable-boundary wrappers), so they participate in checkpointing: on resume from a mid-graph checkpoint, before_run does not re-fire and after_run only fires at real completion. This is the behavior snapshot-style hooks (e.g. MessagePersistenceExtension) depend on, and it falls out naturally from reducer-based state updates. An onion wrap cannot be expressed as a graph node — so rather than adopting divergent semantics that would re-fire on every resume, run-scoped wrapping is omitted. Use before_run + after_run for setup/teardown; use wrap_model or wrap_tool when try/finally-shaped control flow is required.

Sibling-aware configuration via setup()

When an extension needs to react to other extensions in the kit (e.g. enabling a feature only when a particular sibling is present), override setup():

from langchain_agentkit import Extension
from langchain_agentkit.extensions.hitl import HITLExtension

class MyExtension(Extension):
    def __init__(self):
        self._hitl_enabled = False

    def setup(self, *, extensions, **_):
        # Inspect the assembled kit and configure self accordingly.
        self._hitl_enabled = any(isinstance(e, HITLExtension) for e in extensions)

setup() is called once after dependency resolution, before the graph is built. Each extension declares only the kwargs it needs — the framework uses signature introspection to pass only what's requested. Available kwargs:

Contract — inspect presence, not state: setup() runs in declaration order, so another extension's setup() may not have run yet when yours executes. Only inspect sibling presence via isinstance() checks — never read mutable state that another extension's setup() might populate. Anything that depends on a sibling being fully configured should happen lazily at runtime.

Declaring dependencies

If your extension requires another extension to function, declare it via dependencies() — AgentKit will auto-add it if missing:

class MyExtension(Extension):
    def dependencies(self):
        return [TasksExtension()]  # auto-added if user didn't include one

Contributing

git clone https://github.com/rsmdt/langchain-agentkit.git
cd langchain-agentkit
uv sync --extra dev
uv run pytest tests/unit/ -q
uv run ruff check src/ tests/
uv run mypy src/

# LLM integration evals (requires OPENAI_API_KEY in .env)
uv sync --extra eval
uv run pytest tests/evals/ -m eval -v