Langgraph Multiagent

interface:
  display_name: "LangGraph Multiagent"
  short_description: "Design LangGraph multi-agent systems"
  default_prompt: "Use $langgraph-multiagent to plan or review a LangGraph multi-agent architecture."
policy:
  allow_implicit_invocation: false

from __future__ import annotations

import asyncio
import json
import uuid
from dataclasses import asdict, is_dataclass
from typing import Any, AsyncIterator, Literal

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

# NOTE:
# - This template demonstrates in-process LangGraph/LangChain streaming behind FastAPI.
# - For most production UIs, prefer LangGraph Agent Server + useStream() (see references/ui_nextjs_rsc.md).


class ChatMessage(BaseModel):
    role: str
    content: str


class ChatStreamRequest(BaseModel):
    thread_id: str | None = Field(default=None, description="Thread ID for persistence.")
    messages: list[ChatMessage] | None = Field(
        default=None, description="New messages to append (normal run)."
    )
    resume: dict[str, Any] | None = Field(
        default=None,
        description=(
            "Resume payload for Command(resume=...). Must match the interrupt schema you received."
        ),
    )
    stream_modes: list[Literal["messages", "updates", "custom"]] = Field(
        default_factory=lambda: ["messages", "updates"],
        description="Which LangGraph stream modes to consume.",
    )


def _jsonable(value: Any) -> Any:
    if is_dataclass(value):
        return asdict(value)
    if isinstance(value, (str, int, float, bool)) or value is None:
        return value
    if isinstance(value, dict):
        return {str(k): _jsonable(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_jsonable(v) for v in value]
    if isinstance(value, tuple):
        return [_jsonable(v) for v in value]
    return str(value)


def _sse(event: dict[str, Any], *, event_id: int | None = None, name: str | None = None) -> str:
    lines: list[str] = []
    if event_id is not None:
        lines.append(f"id: {event_id}")
    if name is not None:
        lines.append(f"event: {name}")
    lines.append(
        "data: " + json.dumps(event, separators=(",", ":"), default=str)  # type: ignore[arg-type]
    )
    lines.append("")
    return "\n".join(lines) + "\n"


# ---------------------------------------------------------------------------
# Agent wiring (replace with your real multi-agent runtime)
# ---------------------------------------------------------------------------

def build_agent() -> object:
    """
    Replace this with your real agent/graph import.

    For example:
      from src.agent import agent
      return agent
    """
    from langchain.agents import create_agent
    from langchain.chat_models import init_chat_model
    from langchain.tools import tool
    from langgraph.checkpoint.memory import InMemorySaver

    @tool
    def search(query: str) -> str:
        return f"[search results] {query}"

    model = init_chat_model("gpt-4o-mini", temperature=0)
    return create_agent(
        model=model,
        tools=[search],
        system_prompt="You are a helpful assistant. Use tools when useful.",
        checkpointer=InMemorySaver(),
    )


AGENT = build_agent()

app = FastAPI(title="LangGraph SSE (in-process)")


@app.post("/chat/stream")
async def chat_stream(req: ChatStreamRequest) -> StreamingResponse:
    thread_id = req.thread_id or str(uuid.uuid4())

    async def gen() -> AsyncIterator[str]:
        event_id = 0
        yield _sse({"type": "meta", "threadId": thread_id}, event_id=event_id, name="meta")
        event_id += 1

        config = {"configurable": {"thread_id": thread_id}}
        stream_modes = req.stream_modes

        # If resuming after HITL, pass a Command(resume=...) object into the agent.
        if req.resume is not None:
            from langgraph.types import Command

            agent_input: Any = Command(resume=req.resume)
        else:
            if not req.messages:
                yield _sse(
                    {"type": "error", "message": "messages is required when resume is not provided"},
                    event_id=event_id,
                    name="error",
                )
                return
            agent_input = {
                "messages": [m.model_dump() for m in req.messages],
            }

        try:
            # Multi-mode streaming yields (mode, chunk) tuples.
            async for mode, chunk in AGENT.astream(  # type: ignore[attr-defined]
                agent_input,
                config=config,
                stream_mode=stream_modes,
            ):
                if mode == "messages":
                    token, metadata = chunk
                    content = getattr(token, "content", None)
                    if content:
                        meta_json = _jsonable(metadata)
                        yield _sse(
                            {
                                "type": "token",
                                "content": content,
                                "metadata": meta_json,
                                "node": meta_json.get("langgraph_node") if isinstance(meta_json, dict) else None,
                            },
                            event_id=event_id,
                            name="token",
                        )
                        event_id += 1
                elif mode == "updates":
                    data = _jsonable(chunk)
                    yield _sse({"type": "update", "data": data}, event_id=event_id, name="update")
                    event_id += 1

                    # Interrupts often appear under a __interrupt__ key during updates.
                    if isinstance(data, dict) and "__interrupt__" in data:
                        yield _sse(
                            {"type": "interrupt", "interrupt": data["__interrupt__"]},
                            event_id=event_id,
                            name="interrupt",
                        )
                        event_id += 1
                elif mode == "custom":
                    yield _sse(
                        {"type": "custom", "event": _jsonable(chunk)},
                        event_id=event_id,
                        name="custom",
                    )
                    event_id += 1
                else:
                    yield _sse(
                        {"type": "custom", "event": {"mode": mode, "chunk": _jsonable(chunk)}},
                        event_id=event_id,
                        name="custom",
                    )
                    event_id += 1

                await asyncio.sleep(0)
        except Exception as e:  # noqa: BLE001
            yield _sse({"type": "error", "message": str(e)}, event_id=event_id, name="error")
            return

        yield _sse({"type": "done"}, event_id=event_id, name="done")

    headers = {
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
    }
    return StreamingResponse(gen(), media_type="text/event-stream", headers=headers)


# Run with:
#   uvicorn fastapi_sse_multiagent:app --reload --port 8000

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent.py:agent"
  },
  "env": ".env"
}

import { convertToModelMessages, streamText, type UIMessage } from "ai";
import { openai } from "@ai-sdk/openai";

export const maxDuration = 30;

export async function POST(req: Request) {
  const { messages } = (await req.json()) as { messages: UIMessage[] };

  const result = streamText({
    // Swap this for your provider/model.
    model: openai("gpt-4o-mini"),
    system:
      "You are a helpful assistant. Use markdown for clarity when it helps.",
    // v6: convertToModelMessages is async.
    messages: await convertToModelMessages(messages),
  });

  return result.toUIMessageStreamResponse({
    onError: (error) => {
      if (error instanceof Error) return error.message;
      return "Unknown error";
    },
  });
}

"use client";

import { useState } from "react";
import { DefaultChatTransport } from "ai";
import { useChat } from "@ai-sdk/react";

import Response from "./Response";

type Props = {
  api?: string;
};

function safeJson(value: unknown): string {
  try {
    return JSON.stringify(value, null, 2);
  } catch {
    return String(value);
  }
}

export default function ChatClient({ api = "/api/chat" }: Props) {
  const { messages, sendMessage, status, error, stop, regenerate } = useChat({
    transport: new DefaultChatTransport({ api }),
  });

  const [input, setInput] = useState("");

  return (
    <div className="mx-auto flex h-dvh max-w-4xl flex-col gap-4 p-4">
      <header className="flex items-center justify-between gap-2">
        <div className="flex flex-col">
          <div className="text-sm font-medium">AI SDK Chat</div>
          <div className="text-xs text-muted-foreground">
            transport: <span className="font-mono">{api}</span>
          </div>
        </div>
        <div className="flex items-center gap-2">
          <button
            type="button"
            className="rounded-md border px-3 py-1.5 text-sm"
            onClick={() => regenerate()}
            disabled={!(status === "ready" || status === "error")}
          >
            Regenerate
          </button>
          <button
            type="button"
            className="rounded-md border px-3 py-1.5 text-sm"
            onClick={() => stop()}
            disabled={status !== "streaming"}
          >
            Stop
          </button>
        </div>
      </header>

      <main className="flex-1 overflow-auto rounded-lg border bg-background p-3">
        <div className="flex flex-col gap-3">
          {messages.map((message) => (
            <div key={message.id} className="flex flex-col gap-2">
              <div className="text-xs font-semibold uppercase tracking-wide text-muted-foreground">
                {message.role}
              </div>
              <div className="rounded-md border bg-background p-2">
                {message.parts.map((part, idx) => {
                  switch (part.type) {
                    case "text":
                      return (
                        <Response
                          key={idx}
                          isAnimating={status === "streaming"}
                        >
                          {part.text}
                        </Response>
                      );
                    case "tool-invocation":
                      return (
                        <pre
                          key={idx}
                          className="mt-2 overflow-auto rounded bg-muted p-2 font-mono text-[11px]"
                        >
                          {safeJson(part)}
                        </pre>
                      );
                    case "tool-result":
                      return (
                        <pre
                          key={idx}
                          className="mt-2 overflow-auto rounded bg-muted p-2 font-mono text-[11px]"
                        >
                          {safeJson(part)}
                        </pre>
                      );
                    case "file":
                      return (
                        <pre
                          key={idx}
                          className="mt-2 overflow-auto rounded bg-muted p-2 font-mono text-[11px]"
                        >
                          {safeJson(part)}
                        </pre>
                      );
                    default:
                      return (
                        <pre
                          key={idx}
                          className="mt-2 overflow-auto rounded bg-muted p-2 font-mono text-[11px]"
                        >
                          {safeJson(part)}
                        </pre>
                      );
                  }
                })}
              </div>
            </div>
          ))}

          {status === "submitted" || status === "streaming" ? (
            <div className="text-sm text-muted-foreground">Thinking…</div>
          ) : null}
          {error ? (
            <div className="rounded-md border border-red-500/40 bg-red-500/10 p-3 text-sm text-red-500">
              {error.message}
            </div>
          ) : null}
        </div>
      </main>

      <footer className="flex items-end gap-2">
        <textarea
          className="h-20 flex-1 resize-none rounded-md border bg-background p-2 text-sm"
          placeholder="Type a message…"
          value={input}
          onChange={(e) => setInput(e.target.value)}
          onKeyDown={(e) => {
            if (e.key === "Enter" && (e.metaKey || e.ctrlKey)) {
              e.preventDefault();
              void sendMessage({ text: input });
              setInput("");
            }
          }}
        />
        <button
          type="button"
          className="rounded-md border bg-background px-3 py-2 text-sm"
          disabled={status !== "ready"}
          onClick={() => {
            void sendMessage({ text: input });
            setInput("");
          }}
        >
          Send
        </button>
      </footer>
      <div className="text-xs text-muted-foreground">
        Tip: press <span className="font-mono">Ctrl+Enter</span> to send.
      </div>
    </div>
  );
}

import ChatClient from "./ChatClient";

export default function Page() {
  return <ChatClient api="/api/chat" />;
}

"use client";

import { memo, type ComponentProps } from "react";
import { Streamdown } from "streamdown";

type Props = ComponentProps<typeof Streamdown>;

function Response({ className, ...props }: Props) {
  const base = "prose prose-sm dark:prose-invert max-w-none";
  return (
    <Streamdown
      className={className ? `${base} ${className}` : base}
      {...props}
    />
  );
}

export default memo(Response);

"use client";

import { useEffect, useMemo, useState } from "react";
import { useStream } from "@langchain/langgraph-sdk/react";
import type { Message } from "@langchain/langgraph-sdk";

import type { AgentState, HITLRequest, HITLResponse } from "./types";
import { HITLRequestSchema } from "./types";
import Response from "./Response";

type Props = {
  apiUrl: string;
  assistantId: string;
};

type DecisionType = "approve" | "reject" | "edit";
type StreamRef = ReturnType<typeof useStream>;

const THREAD_ID_STORAGE_KEY = "langgraph:thread_id";

function safeStringify(value: unknown): string {
  try {
    return JSON.stringify(value, null, 2);
  } catch {
    return String(value);
  }
}

function renderContent(content: unknown): string {
  if (typeof content === "string") return content;
  if (content == null) return "";
  return safeStringify(content);
}

export default function ChatClient({ apiUrl, assistantId }: Props) {
  const [threadId, setThreadId] = useState<string | null>(null);
  const [input, setInput] = useState("");

  useEffect(() => {
    setThreadId(window.localStorage.getItem(THREAD_ID_STORAGE_KEY));
  }, []);

  useEffect(() => {
    if (!threadId) return;
    window.localStorage.setItem(THREAD_ID_STORAGE_KEY, threadId);
  }, [threadId]);

  const stream = useStream<AgentState, { InterruptType: HITLRequest }>({
    apiUrl,
    assistantId,
    threadId,
    onThreadId: setThreadId,
    reconnectOnMount: true,
  });

  const [hitlUiError, setHitlUiError] = useState<string | null>(null);
  const [decisionModeByIndex, setDecisionModeByIndex] = useState<
    Record<number, DecisionType>
  >({});
  const [editedArgsJsonByIndex, setEditedArgsJsonByIndex] = useState<
    Record<number, string>
  >({});
  const [rejectReason, setRejectReason] = useState("User rejected");

  const hitlRequest = useMemo(() => {
    const candidate = stream.interrupt?.value;
    const parsed = HITLRequestSchema.safeParse(candidate);
    return parsed.success ? parsed.data : undefined;
  }, [stream.interrupt]);

  useEffect(() => {
    if (!hitlRequest) return;
    const nextDecisionModes: Record<number, DecisionType> = {};
    const nextArgs: Record<number, string> = {};
    hitlRequest.actionRequests.forEach((action, idx) => {
      nextDecisionModes[idx] = "approve";
      nextArgs[idx] = safeStringify(action.args ?? action.arguments ?? {});
    });
    setDecisionModeByIndex(nextDecisionModes);
    setEditedArgsJsonByIndex(nextArgs);
    setHitlUiError(null);
  }, [hitlRequest]);

  const handleSubmit = async () => {
    const text = input.trim();
    if (!text) return;
    setInput("");
    await stream.submit({
      messages: [{ type: "human", content: text }],
    });
  };

  const handleNewThread = () => {
    window.localStorage.removeItem(THREAD_ID_STORAGE_KEY);
    setThreadId(null);
    stream.stop();
  };

  const handleResumeFromInterrupt = async () => {
    if (!hitlRequest) return;
    setHitlUiError(null);

    const decisions: HITLResponse["decisions"] = hitlRequest.actionRequests.map(
      (action, idx) => {
        const mode = decisionModeByIndex[idx] ?? "approve";
        if (mode === "approve") return { type: "approve" };
        if (mode === "reject") return { type: "reject", message: rejectReason };

        const raw = editedArgsJsonByIndex[idx] ?? "{}";
        try {
          const args = JSON.parse(raw) as Record<string, unknown>;
          const editedAction =
            action.arguments !== undefined
              ? { name: action.name, arguments: args }
              : { name: action.name, args };
          return {
            type: "edit",
            editedAction,
          };
        } catch {
          return { type: "reject", message: "Invalid edited JSON args" };
        }
      }
    );

    const resume: HITLResponse = { decisions };

    await stream.submit(null, {
      command: { resume },
    });
  };

  return (
    <div className="mx-auto flex h-dvh max-w-4xl flex-col gap-4 p-4">
      <header className="flex items-center justify-between gap-2">
        <div className="flex flex-col">
          <div className="text-sm font-medium">LangGraph Chat</div>
          <div className="text-xs text-muted-foreground">
            assistant: <span className="font-mono">{assistantId}</span> · thread:{" "}
            <span className="font-mono">{threadId ?? "—"}</span>
          </div>
        </div>
        <div className="flex items-center gap-2">
          <button
            type="button"
            className="rounded-md border px-3 py-1.5 text-sm"
            onClick={handleNewThread}
          >
            New thread
          </button>
        </div>
      </header>

      <main className="flex-1 overflow-auto rounded-lg border bg-background p-3">
        <div className="flex flex-col gap-3">
          {stream.messages.map((message, idx) => (
            <MessageRow
              key={(message.id ?? idx) as string}
              message={message as Message}
              stream={stream}
            />
          ))}

          {stream.isLoading ? (
            <div className="text-sm text-muted-foreground">Thinking…</div>
          ) : null}
          {stream.error ? (
            <div className="rounded-md border border-red-500/40 bg-red-500/10 p-3 text-sm text-red-500">
              {stream.error.message}
            </div>
          ) : null}

          {hitlRequest && hitlRequest.actionRequests.length > 0 ? (
            <section className="rounded-lg border border-amber-500/40 bg-amber-500/10 p-3">
              <div className="mb-2 text-sm font-semibold text-amber-600">
                Human approval required
              </div>
              <div className="mb-3 text-xs text-muted-foreground">
                Decide for each action, then resume the run.
              </div>

              <div className="flex flex-col gap-3">
                {hitlRequest.actionRequests.map((action, idx) => (
                  <div
                    key={idx}
                    className="rounded-md border bg-background p-3"
                  >
                    <div className="mb-2 flex items-center justify-between gap-2">
                      <div className="text-sm font-mono">{action.name}</div>
                      <select
                        className="rounded-md border bg-background px-2 py-1 text-xs"
                        value={decisionModeByIndex[idx] ?? "approve"}
                        onChange={(e) =>
                          setDecisionModeByIndex((prev) => ({
                            ...prev,
                            [idx]: e.target.value as DecisionType,
                          }))
                        }
                      >
                        <option value="approve">approve</option>
                        <option value="reject">reject</option>
                        <option value="edit">edit args</option>
                      </select>
                    </div>

                    <div className="mb-2 text-xs text-muted-foreground">
                      {action.description ?? "Tool call pending review."}
                    </div>

                    {(decisionModeByIndex[idx] ?? "approve") === "edit" ? (
                      <textarea
                        className="h-28 w-full rounded-md border bg-background p-2 font-mono text-xs"
                        value={editedArgsJsonByIndex[idx] ?? "{}"}
                        onChange={(e) =>
                          setEditedArgsJsonByIndex((prev) => ({
                            ...prev,
                            [idx]: e.target.value,
                          }))
                        }
                      />
                    ) : (
                      <pre className="overflow-auto rounded-md border bg-muted p-2 text-xs">
                        {safeStringify(action.args ?? action.arguments ?? {})}
                      </pre>
                    )}
                  </div>
                ))}
              </div>

              <div className="mt-3 flex flex-col gap-2">
                <div className="flex items-center gap-2">
                  <input
                    className="w-full rounded-md border bg-background px-2 py-1 text-xs"
                    value={rejectReason}
                    onChange={(e) => setRejectReason(e.target.value)}
                    placeholder="Reject reason (used for any reject decisions)"
                  />
                  <button
                    type="button"
                    className="rounded-md border bg-background px-3 py-1.5 text-xs"
                    onClick={handleResumeFromInterrupt}
                  >
                    Resume
                  </button>
                </div>

                {hitlUiError ? (
                  <div className="text-xs text-red-500">{hitlUiError}</div>
                ) : null}
              </div>
            </section>
          ) : null}
        </div>
      </main>

      <footer className="flex items-end gap-2">
        <textarea
          className="h-20 flex-1 resize-none rounded-md border bg-background p-2 text-sm"
          placeholder="Type a message…"
          value={input}
          onChange={(e) => setInput(e.target.value)}
          onKeyDown={(e) => {
            if (e.key === "Enter" && (e.metaKey || e.ctrlKey)) {
              e.preventDefault();
              handleSubmit();
            }
          }}
        />
        <button
          type="button"
          className="rounded-md border bg-background px-3 py-2 text-sm"
          disabled={stream.isLoading}
          onClick={handleSubmit}
        >
          Send
        </button>
      </footer>
      <div className="text-xs text-muted-foreground">
        Tip: press <span className="font-mono">Ctrl+Enter</span> to send.
      </div>
    </div>
  );
}

function MessageRow({
  message,
  stream,
}: {
  message: Message;
  stream: StreamRef;
}) {
  const meta = stream.getMessageMetadata(message);
  const node = (meta as { langgraph_node?: string } | undefined)?.langgraph_node;
  const role = message.type;
  const toolCalls =
    role === "ai" ? (stream.getToolCalls(message) as unknown[]) : [];
  const content = renderContent(message.content);

  return (
    <div className="flex flex-col gap-2">
      <div className="flex items-center justify-between gap-2">
        <div className="text-xs font-semibold uppercase tracking-wide text-muted-foreground">
          {role}
          {node ? <span className="ml-2 font-mono lowercase">{node}</span> : null}
        </div>
      </div>

      {toolCalls.length > 0 ? (
        <div className="flex flex-col gap-2">
          {toolCalls.map((tc, idx) => (
            <ToolCallCard key={idx} toolCall={tc} />
          ))}
        </div>
      ) : null}

      {role === "ai" && typeof message.content === "string" ? (
        <div className="rounded-md border bg-background p-2">
          <Response isAnimating={stream.isLoading}>{content}</Response>
        </div>
      ) : (
        <pre className="whitespace-pre-wrap rounded-md border bg-muted p-2 text-sm">
          {content}
        </pre>
      )}
    </div>
  );
}

function ToolCallCard({ toolCall }: { toolCall: unknown }) {
  const tc = toolCall as {
    id?: string;
    state?: string;
    call?: { name?: string; args?: unknown };
    result?: unknown;
    error?: unknown;
  };
  return (
    <div className="rounded-md border bg-background p-2 text-xs">
      <div className="flex items-center justify-between gap-2">
        <div className="font-mono">
          {tc.call?.name ?? "tool_call"}
          {tc.id ? <span className="text-muted-foreground"> · {tc.id}</span> : null}
        </div>
        {tc.state ? (
          <div className="rounded bg-muted px-2 py-0.5 font-mono text-[10px]">
            {tc.state}
          </div>
        ) : null}
      </div>
      <pre className="mt-2 overflow-auto rounded bg-muted p-2 font-mono text-[11px]">
        {safeStringify(tc.call?.args ?? {})}
      </pre>
      {tc.result != null ? (
        <pre className="mt-2 overflow-auto rounded bg-muted p-2 font-mono text-[11px]">
          {safeStringify(tc.result)}
        </pre>
      ) : null}
      {tc.error != null ? (
        <pre className="mt-2 overflow-auto rounded bg-red-500/10 p-2 font-mono text-[11px] text-red-500">
          {safeStringify(tc.error)}
        </pre>
      ) : null}
    </div>
  );
}

import ChatClient from "./ChatClient";

export default function Page() {
  const apiUrl =
    process.env.NEXT_PUBLIC_LANGGRAPH_API_URL ?? "http://localhost:2024";
  const assistantId = process.env.NEXT_PUBLIC_LANGGRAPH_ASSISTANT_ID ?? "agent";

  return <ChatClient apiUrl={apiUrl} assistantId={assistantId} />;
}

"use client";

import { memo, type ComponentProps } from "react";
import { Streamdown } from "streamdown";

type Props = ComponentProps<typeof Streamdown>;

function Response({ className, ...props }: Props) {
  const base = "prose prose-sm dark:prose-invert max-w-none";
  return (
    <Streamdown
      className={className ? `${base} ${className}` : base}
      {...props}
    />
  );
}

export default memo(Response);

import { z } from "zod";
import type { Message } from "@langchain/langgraph-sdk";

// ---------------------------------------------------------------------------
// Tool-call typing (fill this in per app)
// ---------------------------------------------------------------------------

export const ToolCallSchema = z.discriminatedUnion("name", [
  z.object({
    name: z.literal("search"),
    args: z.object({ query: z.string() }),
    id: z.string().optional(),
  }),
  z.object({
    name: z.literal("calculate"),
    args: z.object({ expression: z.string() }),
    id: z.string().optional(),
  }),
]);

export type ToolCall = z.infer<typeof ToolCallSchema>;

export type AgentMessage = Message<ToolCall>;

export interface AgentState {
  // Embed tool-call types in messages for end-to-end type safety in useStream().
  messages: AgentMessage[];
  // Optional: keep custom UI messages in graph state for Generative UI.
  ui?: unknown[];
}

// ---------------------------------------------------------------------------
// HITL typing (runtime value is versioned; validate defensively)
// ---------------------------------------------------------------------------

export const HITLActionRequestSchema = z.object({
  name: z.string(),
  // JS docs typically use `args`, but some examples show `arguments`.
  args: z.record(z.unknown()).optional(),
  arguments: z.record(z.unknown()).optional(),
  description: z.string().optional(),
});

export const HITLRequestSchema = z.preprocess(
  (value) => {
    if (value == null || typeof value !== "object") return value;
    const obj = value as Record<string, unknown>;
    return {
      actionRequests: (obj.actionRequests ?? obj.action_requests) as unknown,
      reviewConfigs: (obj.reviewConfigs ?? obj.review_configs) as unknown,
    };
  },
  z.object({
    actionRequests: z.array(HITLActionRequestSchema).default([]),
    reviewConfigs: z.array(z.unknown()).optional(),
  })
);

export type HITLRequest = z.infer<typeof HITLRequestSchema>;

const HITLEditedActionSchema = z.object({
  name: z.string(),
  args: z.record(z.unknown()).optional(),
  arguments: z.record(z.unknown()).optional(),
});

export const HITLDecisionSchema = z.discriminatedUnion("type", [
  z.object({ type: z.literal("approve") }),
  z.object({ type: z.literal("reject"), message: z.string().optional() }),
  z.object({
    type: z.literal("edit"),
    editedAction: HITLEditedActionSchema,
  }),
]);

export const HITLResponseSchema = z.object({
  decisions: z.array(HITLDecisionSchema),
});

export type HITLResponse = z.infer<typeof HITLResponseSchema>;

from __future__ import annotations

import urllib.request

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool


ALLOWED_PREFIXES = (
    "https://langchain-ai.github.io/langgraph/",
    "https://docs.langchain.com/oss/python/",
)


@tool
def fetch_documentation(url: str) -> str:
    """Fetch documentation from an allowlisted URL (SSRF-safe by prefix allowlist)."""
    if not any(url.startswith(p) for p in ALLOWED_PREFIXES):
        return f"Error: URL not allowed. Must start with one of: {', '.join(ALLOWED_PREFIXES)}"
    req = urllib.request.Request(url, headers={"User-Agent": "langgraph-multiagent-skill/1.0"})
    with urllib.request.urlopen(req, timeout=20) as resp:
        return resp.read().decode("utf-8", errors="replace")


def build_docs_agent() -> object:
    model = init_chat_model("gpt-4o-mini", temperature=0)
    system_prompt = (
        "You are a docs-grounded assistant for LangGraph/LangChain.\n"
        "If the question involves API details, you MUST call fetch_documentation\n"
        "on the relevant official docs URL before answering."
    )
    return create_agent(
        model=model,
        tools=[fetch_documentation],
        system_prompt=system_prompt,
    )

from __future__ import annotations

from typing import Literal

from langchain.agents import AgentState, create_agent
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from typing_extensions import NotRequired


class MultiAgentState(AgentState):
    active_agent: NotRequired[str]


@tool
def transfer_to_sales(runtime: ToolRuntime) -> Command:
    """Transfer to the sales agent."""
    last_ai_message = next(msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage))
    transfer_message = ToolMessage(
        content="Transferred to sales agent from support agent",
        tool_call_id=runtime.tool_call_id,
    )
    return Command(
        goto="sales_agent",
        update={"active_agent": "sales_agent", "messages": [last_ai_message, transfer_message]},
        graph=Command.PARENT,
    )


@tool
def transfer_to_support(runtime: ToolRuntime) -> Command:
    """Transfer to the support agent."""
    last_ai_message = next(msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage))
    transfer_message = ToolMessage(
        content="Transferred to support agent from sales agent",
        tool_call_id=runtime.tool_call_id,
    )
    return Command(
        goto="support_agent",
        update={"active_agent": "support_agent", "messages": [last_ai_message, transfer_message]},
        graph=Command.PARENT,
    )


sales_agent = create_agent(
    model="anthropic:claude-sonnet-4-20250514",
    tools=[transfer_to_support],
    system_prompt="You are a sales agent. If asked about support, transfer to support.",
)

support_agent = create_agent(
    model="anthropic:claude-sonnet-4-20250514",
    tools=[transfer_to_sales],
    system_prompt="You are a support agent. If asked about sales, transfer to sales.",
)


def call_sales(state: MultiAgentState) -> Command:
    return sales_agent.invoke(state)


def call_support(state: MultiAgentState) -> Command:
    return support_agent.invoke(state)


def route_after_agent(state: MultiAgentState) -> Literal["sales_agent", "support_agent", END]:
    messages = state.get("messages", [])
    if messages:
        last = messages[-1]
        if isinstance(last, AIMessage) and not last.tool_calls:
            return END
    return state.get("active_agent") or "sales_agent"


def route_initial(state: MultiAgentState) -> Literal["sales_agent", "support_agent"]:
    return state.get("active_agent") or "sales_agent"


def build_graph() -> object:
    builder = StateGraph(MultiAgentState)
    builder.add_node("sales_agent", call_sales)
    builder.add_node("support_agent", call_support)
    builder.add_conditional_edges(START, route_initial, ["sales_agent", "support_agent"])
    builder.add_conditional_edges("sales_agent", route_after_agent, ["sales_agent", "support_agent", END])
    builder.add_conditional_edges("support_agent", route_after_agent, ["sales_agent", "support_agent", END])
    return builder.compile()

from __future__ import annotations

from typing import Literal, TypedDict

from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, MessagesState, StateGraph


class State(MessagesState):
    # Keep additional aggregation state here (lists/dicts) and ensure reducers if parallelizing.
    pass


def build_graph() -> object:
    model = init_chat_model("gpt-4o-mini", temperature=0)

    def plan(state: State) -> dict:
        # Replace with planning logic; keep it small and structured.
        response = model.invoke(state["messages"])
        return {"messages": [response]}

    def should_continue(state: State) -> Literal[END, "plan"]:
        last = state["messages"][-1]
        return END if not getattr(last, "tool_calls", None) else "plan"

    builder = StateGraph(State)
    builder.add_node("plan", plan)
    builder.add_edge(START, "plan")
    builder.add_conditional_edges("plan", should_continue)
    return builder.compile()

from __future__ import annotations

from typing import TypedDict

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver


class Output(TypedDict):
    text: str


@tool
def calendar_tool(request: str) -> str:
    """Placeholder tool: replace with a real calendar integration."""
    return f"[calendar] {request}"


@tool
def email_tool(request: str) -> str:
    """Placeholder tool: replace with a real email integration."""
    return f"[email] {request}"


def build_supervisor_agent() -> object:
    model = init_chat_model("gpt-4o-mini", temperature=0)

    calendar_agent = create_agent(
        model,
        tools=[calendar_tool],
        system_prompt="You are a calendar assistant. Be precise.",
    )

    email_agent = create_agent(
        model,
        tools=[email_tool],
        system_prompt="You are an email assistant. Be concise and professional.",
    )

    @tool
    def schedule_event(request: str) -> str:
        result = calendar_agent.invoke({"messages": [{"role": "user", "content": request}]})
        return result["messages"][-1].text

    @tool
    def manage_email(request: str) -> str:
        result = email_agent.invoke({"messages": [{"role": "user", "content": request}]})
        return result["messages"][-1].text

    supervisor = create_agent(
        model,
        tools=[schedule_event, manage_email],
        system_prompt=(
            "You are a supervisor. Delegate work to tools. "
            "Use multiple tools in sequence when needed."
        ),
        checkpointer=InMemorySaver(),
    )
    return supervisor

from __future__ import annotations

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware, PIIMiddleware, SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver


@tool
def read_data(query: str) -> str:
    return f"[data] {query}"


@tool
def send_email(to: str, subject: str, body: str) -> str:
    return f"Sent email to {to} with subject {subject}"


def build_agent() -> object:
    model = init_chat_model("gpt-4o-mini", temperature=0)

    agent = create_agent(
        model=model,
        tools=[read_data, send_email],
        system_prompt="You are a helpful assistant. Use tools when appropriate.",
        middleware=[
            # Redact emails in user input before sending to model
            PIIMiddleware("email", strategy="redact", apply_to_input=True),
            # Keep long conversations bounded
            SummarizationMiddleware(model=model, trigger={"tokens": 1200}),
            # Require approval for side effects
            HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
        ],
        # Required for interrupts / HITL
        checkpointer=InMemorySaver(),
    )
    return agent

from __future__ import annotations

from dataclasses import dataclass

from langchain.agents import create_agent
from langchain.messages import ToolMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest


@dataclass(frozen=True)
class Context:
    user_id: str
    api_key: str


async def require_authentication(request: MCPToolCallRequest, handler):
    runtime = request.runtime
    if not runtime.state.get("authenticated", False) and request.name in {"delete_file", "export_data"}:
        return ToolMessage(
            content="Authentication required.",
            tool_call_id=runtime.tool_call_id,
        )
    return await handler(request)


async def inject_user_context(request: MCPToolCallRequest, handler):
    runtime = request.runtime
    modified = request.override(args={**request.args, "user_id": runtime.context.user_id})
    return await handler(modified)


async def build_agent() -> object:
    client = MultiServerMCPClient(
        {
            "internal": {"url": "http://localhost:8000/mcp"},
        },
        tool_interceptors=[require_authentication, inject_user_context],
    )
    tools = await client.get_tools()
    return create_agent(
        model="gpt-4o-mini",
        tools=tools,
        context_schema=Context,
    )

from __future__ import annotations

from dataclasses import dataclass

from langchain.tools import ToolRuntime, tool


@dataclass(frozen=True)
class RuntimeContext:
    user_id: str


@tool
def example_tool(runtime: ToolRuntime[RuntimeContext]) -> str:
    # Use runtime.context for DI (user IDs, db clients, etc).
    return f"hello {runtime.context.user_id}"

from __future__ import annotations

import operator
from typing import Annotated
from typing_extensions import TypedDict

from langgraph.graph import START, StateGraph


class State(TypedDict):
    # Append-only list reducer allows parallel nodes to safely write updates.
    completed_sections: Annotated[list[str], operator.add]


def worker_a(state: State) -> dict:
    return {"completed_sections": ["a"]}


def worker_b(state: State) -> dict:
    return {"completed_sections": ["b"]}


def build_graph() -> object:
    g = StateGraph(State)
    g.add_node("a", worker_a)
    g.add_node("b", worker_b)
    g.add_edge(START, "a")
    g.add_edge(START, "b")
    return g.compile()

from __future__ import annotations

from graph_supervisor_subagents import build_supervisor_agent


def test_supervisor_smoke() -> None:
    agent = build_supervisor_agent()
    out = agent.invoke({"messages": [{"role": "user", "content": "Draft an email to Bob about lunch."}]})
    assert out["messages"]

from __future__ import annotations

import json
import os
import uuid
from dataclasses import asdict, is_dataclass
from typing import Any

import streamlit as st


def _jsonable(value: Any) -> Any:
    if is_dataclass(value):
        return asdict(value)
    if isinstance(value, (str, int, float, bool)) or value is None:
        return value
    if isinstance(value, dict):
        return {str(k): _jsonable(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_jsonable(v) for v in value]
    if isinstance(value, tuple):
        return [_jsonable(v) for v in value]
    return str(value)


@st.cache_resource
def build_agent() -> object:
    """
    Replace with your real multi-agent system.

    For example:
      from src.agent import agent
      return agent
    """
    if os.environ.get("LANGGRAPH_UI_TEST_MODE") == "1":
        class _Token:
            def __init__(self, content: str) -> None:
                self.content = content

        class _StubAgent:
            def stream(self, agent_input: Any, *_args: Any, **_kwargs: Any):  # type: ignore[no-untyped-def]
                user_text = ""
                if isinstance(agent_input, dict):
                    messages = agent_input.get("messages") or []
                    if messages:
                        last = messages[-1]
                        if isinstance(last, dict):
                            user_text = str(last.get("content") or "")

                response = f"Stub response (set LANGGRAPH_UI_TEST_MODE=0 for real LLM): {user_text}"
                for token in response.split(" "):
                    yield ("messages", (_Token(token + " "), {}))

        return _StubAgent()

    from langchain.agents import create_agent
    from langchain.agents.middleware import HumanInTheLoopMiddleware
    from langchain.chat_models import init_chat_model
    from langchain.tools import tool
    from langgraph.checkpoint.memory import InMemorySaver

    @tool
    def search(query: str) -> str:
        return f"[search results] {query}"

    model = init_chat_model("gpt-4o-mini", temperature=0)
    return create_agent(
        model=model,
        tools=[search],
        system_prompt="You are a helpful assistant. Use tools when useful.",
        middleware=[
            # Example: require approvals for a specific tool (update to match your tools).
            HumanInTheLoopMiddleware(
                interrupt_on={"search": True},
                description_prefix="Tool execution pending approval",
            ),
        ],
        checkpointer=InMemorySaver(),
    )


AGENT = build_agent()

st.set_page_config(page_title="LangGraph Multi-Agent (Streamlit)", layout="wide")


def init_state() -> None:
    st.session_state.setdefault("thread_id", str(uuid.uuid4()))
    st.session_state.setdefault("messages", [])
    st.session_state.setdefault("pending_interrupts", None)
    st.session_state.setdefault("reject_reason", "User rejected")


def reset_thread() -> None:
    st.session_state.thread_id = str(uuid.uuid4())
    st.session_state.messages = []
    st.session_state.pending_interrupts = None


def extract_action_requests(interrupt_value: Any) -> list[dict[str, Any]]:
    if isinstance(interrupt_value, dict):
        if "action_requests" in interrupt_value:
            return interrupt_value.get("action_requests") or []
        if "actionRequests" in interrupt_value:
            return interrupt_value.get("actionRequests") or []
    return []


def stream_run(agent_input: Any) -> None:
    config = {"configurable": {"thread_id": st.session_state.thread_id}}

    assistant_container = st.chat_message("assistant")
    text_placeholder = assistant_container.empty()
    token_buffer = ""
    interrupts: Any = None

    for mode, chunk in AGENT.stream(  # type: ignore[attr-defined]
        agent_input,
        config=config,
        stream_mode=["messages", "updates"],
    ):
        if mode == "messages":
            token, _meta = chunk
            content = getattr(token, "content", None)
            if content:
                token_buffer += content
                text_placeholder.markdown(token_buffer)
        elif mode == "updates":
            data = _jsonable(chunk)
            if isinstance(data, dict) and "__interrupt__" in data:
                interrupts = data["__interrupt__"]

    if token_buffer.strip():
        st.session_state.messages.append({"role": "assistant", "content": token_buffer})

    st.session_state.pending_interrupts = interrupts


def resume_from_interrupts(decisions: dict[str, Any]) -> None:
    from langgraph.types import Command

    stream_run(Command(resume=decisions))


init_state()

st.sidebar.header("Session")
st.sidebar.caption(f"Streamlit {st.__version__}")
st.sidebar.caption("Thread IDs are required for persistence, interrupts, and resume.")
st.sidebar.code(st.session_state.thread_id, language="text")

if st.sidebar.button("New thread", use_container_width=True):
    reset_thread()
    st.rerun()

st.title("LangGraph Multi-Agent (Streamlit)")
st.caption("Streams tokens + handles interrupts (HITL) using thread-scoped persistence.")

for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

user_text = st.chat_input("Ask something…")
if user_text:
    st.session_state.messages.append({"role": "user", "content": user_text})
    with st.chat_message("user"):
        st.markdown(user_text)
    stream_run({"messages": [{"role": "user", "content": user_text}]})

pending = st.session_state.pending_interrupts
if pending:
    st.divider()
    st.subheader("Human approval required")

    # In Python, interrupts are commonly a list of Interrupt objects with {id, value}.
    pending_json = _jsonable(pending)
    st.caption("Review each tool call. Decisions must match the action order.")

    reject_reason = st.text_input(
        "Reject reason (used for any reject decisions)",
        value=st.session_state.reject_reason,
        key="hitl_reject_reason",
    )
    st.session_state.reject_reason = reject_reason

    interrupts_list = pending_json if isinstance(pending_json, list) else [pending_json]
    for item in interrupts_list:
        interrupt_id = (item.get("id") if isinstance(item, dict) else None) or str(uuid.uuid4())
        value = item.get("value") if isinstance(item, dict) else item
        action_requests = extract_action_requests(value)

        st.markdown(f"**Interrupt** `{interrupt_id}`")
        st.code(json.dumps(_jsonable(value), indent=2), language="json")

        for idx, action in enumerate(action_requests):
            name = action.get("name") if isinstance(action, dict) else "tool_call"
            args = {}
            if isinstance(action, dict):
                args = action.get("args") or action.get("arguments") or {}

            st.markdown(f"**{name}**")

            decision_key = f"hitl_decision::{interrupt_id}::{idx}"
            mode = st.selectbox(
                "Decision",
                options=["approve", "reject", "edit"],
                index=0,
                key=decision_key,
            )

            if mode == "edit":
                st.text_area(
                    "Edited args (JSON)",
                    value=json.dumps(args, indent=2),
                    key=f"hitl_edit_args::{interrupt_id}::{idx}",
                    height=120,
                )
            else:
                st.code(json.dumps(args, indent=2), language="json")

            st.divider()

    if st.button("Resume", type="primary"):
        resume: dict[str, Any] = {}
        errors: list[str] = []

        for item in interrupts_list:
            interrupt_id = (item.get("id") if isinstance(item, dict) else None) or str(uuid.uuid4())
            value = item.get("value") if isinstance(item, dict) else item
            action_requests = extract_action_requests(value)

            decisions: list[dict[str, Any]] = []
            for idx, action in enumerate(action_requests):
                name = action.get("name") if isinstance(action, dict) else "tool_call"

                mode = st.session_state.get(f"hitl_decision::{interrupt_id}::{idx}", "approve")
                if mode == "approve":
                    decisions.append({"type": "approve"})
                elif mode == "reject":
                    decisions.append({"type": "reject", "message": st.session_state.reject_reason})
                else:
                    raw = st.session_state.get(f"hitl_edit_args::{interrupt_id}::{idx}", "{}")
                    try:
                        edited_args = json.loads(raw)
                    except Exception:
                        errors.append(f"Invalid JSON for edited args: interrupt={interrupt_id} action={idx}")
                        edited_args = {}
                    decisions.append(
                        {
                            "type": "edit",
                            # Python uses snake_case for decision payload keys.
                            "edited_action": {"name": name, "args": edited_args},
                        }
                    )

            resume[interrupt_id] = {"decisions": decisions}

        if errors:
            st.error("\n".join(errors))
        else:
            st.session_state.pending_interrupts = None
            resume_from_interrupts(resume)
            st.rerun()

import os

from streamlit.testing.v1 import AppTest


def test_streamlit_app_loads_smoke() -> None:
    # Keep tests offline; the template supports a stub runtime in test mode.
    os.environ["LANGGRAPH_UI_TEST_MODE"] = "1"

    # NOTE: Adjust the path if you rename/move the Streamlit entrypoint.
    AppTest.from_file("streamlit_chat_app.py").run()