| #!/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)" |