# Script for creating agents which can plug anywhere

> Source: <https://gist.github.com/kevinrawal/5b7b93213bbe7d50a2c3ac295b3fd5d4>
> Published: 2026-08-03 06:58:54+00:00

| #!/usr/bin/env bash | |
| # create-agent.sh — scaffold a portable, embeddable LangGraph agent codebase. | |
| # Fully generic: no domain-specific nodes, works for ANY agent you build. | |
| # | |
| # Usage: | |
| # ./create-agent.sh my_agent | |
| # | |
| # After scaffolding, it installs deps and RUNS the agent immediately so | |
| # you get proof-of-life output, not just a pile of files. | |
| set -euo pipefail | |
| if [ -z "${1:-}" ]; then | |
| echo "Usage: $0 <agent_name_snake_case>" | |
| exit 1 | |
| fi | |
| NAME="$1" | |
| PKG="src/${NAME}" | |
| mkdir -p "${PKG}"/{nodes,tools,adapters} | |
| mkdir -p examples tests | |
| # ---------- state.py ---------- | |
| cat > "${PKG}/state.py" << 'EOF' | |
| """ | |
| Public contract for this agent. Treat like a versioned API schema: | |
| additive changes only, since hosts embedding this agent depend on these shapes. | |
| """ | |
| from typing import TypedDict, Optional | |
| class InputState(TypedDict): | |
| """What a caller must provide. Add/rename fields for your agent.""" | |
| query: str | |
| context: Optional[dict] | |
| class OutputState(TypedDict): | |
| """What a caller gets back. Nothing else leaks out of the graph.""" | |
| result: str | |
| metadata: dict | |
| class InternalState(InputState, OutputState): | |
| """Full scratchpad state used only inside this graph's nodes. | |
| Add whatever fields your nodes need to pass data between steps.""" | |
| iterations: int | |
| scratch: dict | |
| EOF | |
| # ---------- config.py ---------- | |
| cat > "${PKG}/config.py" << 'EOF' | |
| """ | |
| Everything that varies by host (LLM client, checkpointer, tools, callbacks) | |
| is injected here, never imported directly inside node files. | |
| """ | |
| from dataclasses import dataclass, field | |
| from typing import Any, Callable, Optional | |
| @dataclass | |
| class AgentConfig: | |
| llm: Any = None # inject a real client for production use | |
| checkpointer: Optional[Any] = None # host may pass Postgres/Redis saver; None = in-memory | |
| tools: list = field(default_factory=list) | |
| max_iterations: int = 5 | |
| on_event: Optional[Callable[[str, dict], None]] = None # optional observability hook | |
| class MockLLM: | |
| """Zero-setup stand-in so the scaffold runs immediately with no API keys. | |
| Swap AgentConfig.llm for a real client (ChatAnthropic, ChatOpenAI, etc.).""" | |
| def invoke(self, prompt: str) -> str: | |
| return f"[mock output for: {str(prompt)[:60]}]" | |
| def default_config() -> AgentConfig: | |
| """Sensible standalone defaults — used by examples/ and the self-test below.""" | |
| return AgentConfig(llm=MockLLM()) | |
| EOF | |
| # ---------- graph.py ---------- | |
| cat > "${PKG}/graph.py" << EOF | |
| """ | |
| The single product of this package: build_graph(config) -> CompiledStateGraph. | |
| Every adapter (subgraph / tool / MCP / REST / CLI) wraps THIS and only this. | |
| The example flow below (process -> loop -> finish) is a placeholder. | |
| Replace the nodes in nodes/ with your real logic; the graph wiring here | |
| is deliberately generic so it fits any agent shape you build. | |
| """ | |
| from langgraph.graph import StateGraph, START, END | |
| from langgraph.checkpoint.memory import MemorySaver | |
| from .state import InternalState, InputState, OutputState | |
| from .config import AgentConfig | |
| from .nodes.process import make_process_node | |
| from .nodes.finish import make_finish_node | |
| def should_continue(state: InternalState) -> str: | |
| if state.get("iterations", 0) >= state.get("_max_iterations", 5): | |
| return "finish" | |
| return "finish" if state.get("scratch", {}).get("done", True) else "process" | |
| def build_graph(config: AgentConfig): | |
| builder = StateGraph(InternalState, input_schema=InputState, output_schema=OutputState) | |
| builder.add_node("process", make_process_node(config)) | |
| builder.add_node("finish", make_finish_node(config)) | |
| builder.add_edge(START, "process") | |
| builder.add_conditional_edges("process", should_continue, { | |
| "process": "process", | |
| "finish": "finish", | |
| }) | |
| builder.add_edge("finish", END) | |
| checkpointer = config.checkpointer or MemorySaver() | |
| return builder.compile(checkpointer=checkpointer) | |
| EOF | |
| # ---------- nodes/process.py (generic — replace with your logic) ---------- | |
| cat > "${PKG}/nodes/process.py" << 'EOF' | |
| """Generic processing node. Factory pattern: takes AgentConfig, returns a | |
| node function. Replace the body with your actual step logic — this is | |
| just a working placeholder so the graph runs out of the box.""" | |
| from ..config import AgentConfig | |
| from ..state import InternalState | |
| def make_process_node(config: AgentConfig): | |
| def _node(state: InternalState) -> dict: | |
| iterations = state.get("iterations", 0) + 1 | |
| if config.on_event: | |
| config.on_event("process", {"iteration": iterations}) | |
| # TODO: real logic here, e.g. call config.llm.invoke(...) or config.tools | |
| return { | |
| "iterations": iterations, | |
| "scratch": {"done": True}, # flip logic here to loop more than once | |
| } | |
| return _node | |
| EOF | |
| # ---------- nodes/finish.py (generic — replace with your logic) ---------- | |
| cat > "${PKG}/nodes/finish.py" << 'EOF' | |
| """Generic finishing node. Produces the OutputState. Replace with your | |
| actual synthesis/response-formatting logic.""" | |
| from ..config import AgentConfig | |
| from ..state import InternalState | |
| def make_finish_node(config: AgentConfig): | |
| def _node(state: InternalState) -> dict: | |
| query = state.get("query", "") | |
| output = config.llm.invoke(query) if config.llm else f"processed: {query}" | |
| return { | |
| "result": output, | |
| "metadata": {"iterations": state.get("iterations", 0)}, | |
| } | |
| return _node | |
| EOF | |
| # ---------- tools/registry.py ---------- | |
| cat > "${PKG}/tools/registry.py" << 'EOF' | |
| """Central place to define/collect tools. Hosts can override via | |
| AgentConfig.tools instead of editing this file, so the agent stays embeddable.""" | |
| from langchain_core.tools import tool | |
| @tool | |
| def example_tool(query: str) -> str: | |
| """Placeholder tool. Replace or extend, or inject alternatives via config.""" | |
| return f"result for: {query}" | |
| DEFAULT_TOOLS = [example_tool] | |
| EOF | |
| # ---------- adapters/as_subgraph.py ---------- | |
| cat > "${PKG}/adapters/as_subgraph.py" << EOF | |
| """ | |
| Embed this agent as a single node inside a HOST LangGraph graph. | |
| from ${NAME}.adapters.as_subgraph import get_subgraph_node | |
| parent_builder.add_node("my_agent", get_subgraph_node(config)) | |
| """ | |
| from ..graph import build_graph | |
| from ..config import AgentConfig, default_config | |
| def get_subgraph_node(config: AgentConfig = None): | |
| """Returns the compiled graph directly — LangGraph treats a compiled | |
| StateGraph as a valid node. Map field names in the host graph if the | |
| parent's state keys differ from InputState/OutputState.""" | |
| return build_graph(config or default_config()) | |
| EOF | |
| # ---------- adapters/as_tool.py ---------- | |
| cat > "${PKG}/adapters/as_tool.py" << EOF | |
| """ | |
| Expose this agent as a single LangChain tool for a host ReAct-style agent | |
| that wants to call it as a leaf action rather than embed it as a subgraph. | |
| """ | |
| from langchain_core.tools import tool | |
| from ..graph import build_graph | |
| from ..config import AgentConfig, default_config | |
| def get_agent_tool(config: AgentConfig = None): | |
| graph = build_graph(config or default_config()) | |
| @tool | |
| def ${NAME}(query: str) -> str: | |
| """Run the ${NAME} agent on a query and return its result.""" | |
| out = graph.invoke({"query": query, "context": None}, | |
| config={"configurable": {"thread_id": "1"}}) | |
| return out["result"] | |
| return ${NAME} | |
| EOF | |
| # ---------- adapters/as_mcp_server.py ---------- | |
| cat > "${PKG}/adapters/as_mcp_server.py" << EOF | |
| """ | |
| Expose this agent as an MCP server so any MCP-capable host (agentic IDE, | |
| Claude Code, Cursor, etc.) can call it as a single tool without custom | |
| integration code. Run: python -m ${NAME}.adapters.as_mcp_server | |
| """ | |
| from mcp.server.fastmcp import FastMCP | |
| from ..graph import build_graph | |
| from ..config import default_config | |
| mcp = FastMCP("${NAME}") | |
| @mcp.tool() | |
| def run_agent(query: str, context: dict | None = None) -> dict: | |
| """Run the agent and return its OutputState.""" | |
| graph = build_graph(default_config()) # swap in real config/llm as needed | |
| return graph.invoke({"query": query, "context": context}, | |
| config={"configurable": {"thread_id": "1"}}) | |
| if __name__ == "__main__": | |
| mcp.run() | |
| EOF | |
| # ---------- adapters/as_rest_api.py ---------- | |
| cat > "${PKG}/adapters/as_rest_api.py" << EOF | |
| """ | |
| Expose this agent over HTTP for language-agnostic hosts. | |
| Run: uvicorn ${NAME}.adapters.as_rest_api:app --reload | |
| """ | |
| from fastapi import FastAPI | |
| from pydantic import BaseModel | |
| from ..graph import build_graph | |
| from ..config import default_config | |
| app = FastAPI(title="${NAME}") | |
| _graph = build_graph(default_config()) # wire real config at startup | |
| class Request(BaseModel): | |
| query: str | |
| context: dict | None = None | |
| @app.post("/invoke") | |
| def invoke(req: Request): | |
| return _graph.invoke({"query": req.query, "context": req.context}, | |
| config={"configurable": {"thread_id": "1"}}) | |
| EOF | |
| # ---------- adapters/as_cli.py ---------- | |
| cat > "${PKG}/adapters/as_cli.py" << EOF | |
| """Local dev/debug entrypoint. Run: python -m ${NAME}.adapters.as_cli "your query" """ | |
| import sys | |
| from ..graph import build_graph | |
| from ..config import default_config | |
| def main(): | |
| query = " ".join(sys.argv[1:]) or "example query" | |
| graph = build_graph(default_config()) | |
| result = graph.invoke({"query": query, "context": None}, | |
| config={"configurable": {"thread_id": "1"}}) | |
| print(result) | |
| if __name__ == "__main__": | |
| main() | |
| EOF | |
| # ---------- __init__.py files ---------- | |
| touch "${PKG}/__init__.py" "${PKG}/nodes/__init__.py" "${PKG}/tools/__init__.py" "${PKG}/adapters/__init__.py" | |
| # ---------- examples ---------- | |
| cat > "examples/standalone_run.py" << EOF | |
| from ${NAME}.graph import build_graph | |
| from ${NAME}.config import default_config | |
| if __name__ == "__main__": | |
| graph = build_graph(default_config()) | |
| result = graph.invoke({"query": "example question", "context": None}, | |
| config={"configurable": {"thread_id": "1"}}) | |
| print(result) | |
| EOF | |
| cat > "examples/embed_in_parent_graph.py" << EOF | |
| """Shows how a HOST system with its own LangGraph would embed this agent as one node.""" | |
| from langgraph.graph import StateGraph, START, END | |
| from typing import TypedDict | |
| from ${NAME}.adapters.as_subgraph import get_subgraph_node | |
| class HostState(TypedDict): | |
| query: str | |
| context: dict | None | |
| result: str | |
| metadata: dict | |
| host_builder = StateGraph(HostState) | |
| host_builder.add_node("${NAME}", get_subgraph_node()) | |
| host_builder.add_edge(START, "${NAME}") | |
| host_builder.add_edge("${NAME}", END) | |
| host_graph = host_builder.compile() | |
| if __name__ == "__main__": | |
| print(host_graph.invoke({"query": "example question", "context": None}, | |
| config={"configurable": {"thread_id": "1"}})) | |
| EOF | |
| # ---------- tests ---------- | |
| cat > "tests/test_graph.py" << EOF | |
| from ${NAME}.graph import build_graph | |
| from ${NAME}.config import default_config | |
| def test_graph_compiles(): | |
| graph = build_graph(default_config()) | |
| assert graph is not None | |
| def test_graph_runs_end_to_end(): | |
| graph = build_graph(default_config()) | |
| out = graph.invoke({"query": "hello", "context": None}, | |
| config={"configurable": {"thread_id": "1"}}) | |
| assert "result" in out | |
| EOF | |
| # ---------- pyproject.toml ---------- | |
| cat > "pyproject.toml" << EOF | |
| [project] | |
| name = "${NAME}" | |
| version = "0.1.0" | |
| requires-python = ">=3.11" | |
| dependencies = [ | |
| "langgraph>=0.2", | |
| "langchain-core", | |
| "fastapi", | |
| "uvicorn", | |
| "mcp", | |
| ] | |
| [tool.setuptools.packages.find] | |
| where = ["src"] | |
| EOF | |
| # ---------- README.md ---------- | |
| cat > "README.md" << EOF | |
| # ${NAME} | |
| Single source of truth: \`src/${NAME}/graph.py\` — \`build_graph(config) -> CompiledStateGraph\`. | |
| Everything else is an adapter around it. Replace the placeholder nodes in | |
| \`nodes/\` with your real logic; do NOT add invocation-layer imports | |
| (FastAPI, MCP SDK, etc.) inside \`nodes/\`, \` graph.py\`, or \` state.py\` — | |
| that belongs only in \`adapters/\`. | |
| ## Structure | |
| - \`state.py\` — public contract (InputState/OutputState) + internal scratchpad. | |
| - \`config.py\` — injected LLM/checkpointer/tools/callbacks. MockLLM by default so it runs with zero setup. | |
| - \`nodes/\` — one file per graph node, factory pattern \`make_x_node(config)\`. Replace with real logic. | |
| - \`tools/registry.py\` — default tools; hosts can override via \`AgentConfig.tools\`. | |
| - \`adapters/\` — every way this agent can be consumed: | |
| - \`as_subgraph.py\` — embed as a node in another LangGraph graph | |
| - \`as_tool.py\` — expose as a single LangChain tool for a ReAct-style host | |
| - \`as_mcp_server.py\` — expose via MCP for agentic IDEs / Claude Code-like hosts | |
| - \`as_rest_api.py\` — expose over HTTP for non-Python hosts | |
| - \`as_cli.py\` — local dev/debug entrypoint | |
| ## Quickstart | |
| \`\`\` bash | |
| pip install -e . | |
| python examples/standalone_run.py | |
| python -m ${NAME}.adapters.as_cli "your query" | |
| \`\`\` | |
| EOF | |
| echo "✅ Scaffolded '${NAME}'." | |
| echo "" | |
| # Only install what's missing — don't touch existing system packages. | |
| echo "Checking dependencies..." | |
| python3 -c "import langgraph, langchain_core" 2>/dev/null || \ | |
| pip install --break-system-packages -q langgraph langchain-core | |
| echo "" | |
| echo "=== Self-test: running the agent right now (no install needed) ===" | |
| PYTHONPATH="src:${PYTHONPATH:-}" python3 examples/standalone_run.py | |
| echo "=== If you see a result dict above, the scaffold works. ===" | |
| echo "" | |
| echo "To use it as an installable package later: pip install -e . (in a venv)" |