CopilotKit 源码分析

CopilotKit 整体架构总览

前端组件如何注册到后端agent服务中的

通过CopilotKit 生成式UI框架 了解到，根据开放程度可以分成3类组件

1. controrolled Generative UI
2. declarative Generative UI
3. open Generative UI

其中第2和3类都是在CopilotRuntime 中注册的，注册后在后续发给agent的http请求中会以上下文的方式(其实也是参数)传递给agent

onst langGraphAgent = new LangGraphHttpAgent({ url: "http://localhost:8004" });

const runtime = new CopilotRuntime({
  agents: { default: langGraphAgent },
  a2ui: { injectA2UITool: true }, // 启用 declarative Generative UI
  openGenerativeUI: true, // 启用 open Generative UI  代理可以生成任意类型的用户界面——包括 HTML、CSS、JavaScript 等代码，并可以直接在聊天界面中展示
  mcpApps: { // 注册 MCP 应用程序
    servers: [
      {
        type: "http",
        url: "https://mcp.excalidraw.com", // <- Exalidraw MCP Server
        serverId: "example_mcp_server",
      },
    ],
  },
});

const app = createCopilotEndpoint({
  runtime,
  basePath: "/api/copilotkit",
});

serve({ fetch: app.fetch, port: 4004 }, () => {
  console.log("\u2713 CopilotKit API server running at http://localhost:4004");
});

三类组件注册流程总览

源码分析：
已知CopilotKit 是一个基于agui协议的框架,agui协议是基于http的上层协议
LangGraphHttpAgent 从httpAgent 中继承
本身的作用就是通过http协议与agent进行通信

The HttpAgent extends AbstractAgent to provide HTTP-based connectivity to remote AI agents. It handles the request/response cycle and transforms the HTTP event stream into standard Agent User Interaction Protocol events.

CopilotRuntime 是一个桥接器，用于前端和后端agent服务之间的通信，控制请求发送，信息处理等操作

createCopilotEndpoint 会创建一个node服务，利用CopilotRuntime实例，将tools 信息给到agent 并处理请求过程中的信息，返回给前端

对于第1类 使用 useComponent 进行组件注册的过程会使用useFrontendTool 方法，将组件注册到copilotkit 实例的tools 中

对于 使用 组件传入的tools 在实例化过程中会自动注册到copilotkit 实例的tools 中

从输入框提交信息后会经过copiloykit 实例发起runAgent

在初始化创建的时候就会传入tools, 后续在connectAgent 和 runAgent 中都会将tools 信息传递给后端agent

agent 本身是一个http-agent

var ProxiedCopilotRuntimeAgent = class ProxiedCopilotRuntimeAgent extends HttpAgent {
  constructor(options) {
    super(options);
  }
}

多样的UI界面是谁渲染的

UI 渲染总览流程

UI 的渲染依然通过前端组件渲染，只不过在渲染时会根据messages 中的返回信息决定使用组件的类型，选择不同的组件进行渲染。
因此, agent 会决定渲染的组件类型，前端组件只是根据组件类型进行渲染。

源码分析：

 const BoundMessageView = renderSlot(messageView, CopilotChatMessageView, {
    messages,
    isRunning,
  });
  ...
    if (children) {
    return (
      <div data-copilotkit style={{ display: "contents" }}>
        {children({
          messageView: BoundMessageView,
          input: BoundInput,
          scrollView: BoundScrollView,
          suggestionView: BoundSuggestionView ?? <></>,
        })}
      </div>
    );
  }

------------------------------CopilotChatMessageView---------------------------------------------

 // Build the flat element list only when we're not virtualizing (avoids
  // creating 500 React elements that we'd immediately discard).
  const messageElements: React.ReactElement[] = shouldVirtualize
    ? []
    : deduplicatedMessages.flatMap(renderMessageBlock);

  // ---------------------------------------------------------------------------
  // children render prop (custom layout, always non-virtual)
  // ---------------------------------------------------------------------------
  if (children) {
    return (
      <div data-copilotkit style={{ display: "contents" }}>
        {children({ messageElements, messages, isRunning, interruptElement })}
      </div>
    );
  }

renderMessageBlock 是一个函数，用于根据消息的类型渲染不同的组件

controlled Generative UI 渲染逻辑

第一类渲染时走 role: “assistant”

const { Component: AssistantComponent, slotProps: assistantSlotProps } =
useMemo(
    () => resolveSlotComponent(assistantMessage, CopilotChatAssistantMessage),
    [assistantMessage],
);

-----------------------------CopilotChatAssistantMessage--------------------------

const boundToolCallsView = renderSlot(
    toolCallsView,
    CopilotChatToolCallsView,
    {
      message,
      messages,
    },
  );

export function CopilotChatToolCallsView({
  message,
  messages = [],
}: CopilotChatToolCallsViewProps) {
  const renderToolCall = useRenderToolCall();

  if (!message.toolCalls || message.toolCalls.length === 0) {
    return null;
  }

  return (
    <>
      {message.toolCalls.map((toolCall) => {
        const toolMessage = messages.find(
          (m) => m.role === "tool" && m.toolCallId === toolCall.id,
        ) as ToolMessage | undefined;

        return (
          <React.Fragment key={toolCall.id}>
            {renderToolCall({
              toolCall,
              toolMessage,
            })}
          </React.Fragment>
        );
      })}
    </>
  );
}

const renderToolCall = useCallback(
    ({
      toolCall,
      toolMessage,
    }: UseRenderToolCallProps): React.ReactElement | null => {
      // Find the render config for this tool call by name
      // For rendering, we show all tool calls regardless of agentId
      // The agentId scoping only affects handler execution (in core)
      // Priority order:
      // 1. Exact match by name (prefer agent-specific if multiple exist)
      // 2. Wildcard (*) renderer
      const exactMatches = renderToolCalls.filter(
        (rc) => rc.name === toolCall.function.name,
      );

      // If multiple renderers with same name exist, prefer the one matching our agentId
      const renderConfig =
        exactMatches.find((rc) => rc.agentId === agentId) ||
        exactMatches.find((rc) => !rc.agentId) ||
        exactMatches[0] ||
        renderToolCalls.find((rc) => rc.name === "*");

      if (!renderConfig) {
        return null;
      }

      const RenderComponent = renderConfig.render;
      const isExecuting = executingToolCallIds.has(toolCall.id);

      // Use the memoized ToolCallRenderer component to prevent unnecessary re-renders
      return (
        <ToolCallRenderer
          key={toolCall.id}
          toolCall={toolCall}
          toolMessage={toolMessage}
          RenderComponent={RenderComponent}
          isExecuting={isExecuting}
        />
      );
    },
    [renderToolCalls, executingToolCallIds, agentId],
  );

const ToolCallRenderer = React.memo(
  function ToolCallRenderer({
    toolCall,
    toolMessage,
    RenderComponent,
    isExecuting,
  }: ToolCallRendererProps) {
    // Memoize args based on the arguments string to maintain stable reference
    const args = useMemo(
      () => partialJSONParse(toolCall.function.arguments),
      [toolCall.function.arguments],
    );

    const toolName = toolCall.function.name;

    // Render based on status to preserve discriminated union type inference
    if (toolMessage) {
      return (
        <RenderComponent
          name={toolName}
          args={args}
          status={ToolCallStatus.Complete}
          result={toolMessage.content}
        />
      );
    } else if (isExecuting) {
      return (
        <RenderComponent
          name={toolName}
          args={args}
          status={ToolCallStatus.Executing}
          result={undefined}
        />
      );
    } else {
      return (
        <RenderComponent
          name={toolName}
          args={args}
          status={ToolCallStatus.InProgress}
          result={undefined}
        />
      );
    }
  },
  // Custom comparison function to prevent re-renders when tool call data hasn't changed
  ....
)

注意上面的RenderComponent

<RenderComponent
   name={toolName}
   args={args}
   status={ToolCallStatus.Complete}
   result={toolMessage.content}
   />

是经过useFrontendTool 注册的组件是的render

非controlled Generative UI 渲染逻辑

第二类和第三类的渲染逻辑 走 role: “activity” 的渲染逻辑

const { renderActivityMessage } = useRenderActivityMessage();

 ----------------------- useRenderActivityMessage --------------------------------

const { copilotkit } = useCopilotKit();
const config = useCopilotChatConfiguration();
const agentId = config?.agentId ?? DEFAULT_AGENT_ID;

const renderers = copilotkit.renderActivityMessages;

// Find the renderer for a given activity type
  const findRenderer = useCallback(
    (activityType: string): ReactActivityMessageRenderer<unknown> | null => {
      if (!renderers.length) {
        return null;
      }

      const matches = renderers.filter(
        (renderer) => renderer.activityType === activityType,
      );

      return (
        matches.find((candidate) => candidate.agentId === agentId) ??
        matches.find((candidate) => candidate.agentId === undefined) ??
        renderers.find((candidate) => candidate.activityType === "*") ??
        null
      );
    },
    [agentId, renderers],
  );

注册时的处理第2类和第3类 的render

a2UI

dynamic schema UI 渲染messages

fixed schema UI 渲染messages

/**
 * Renders a single A2UI surface using the React renderer.
 * Wraps A2UIProvider + A2UIRenderer and bridges actions back to CopilotKit.
 */
function ReactSurfaceHost({
  surfaceId,
  operations,
  theme,
  agent,
  copilotkit,
  catalog,
}: ReactSurfaceHostProps) {
  // Bridge: when the React renderer dispatches an action, forward to CopilotKit
  const handleAction = useCallback(
    async (message: A2UIClientEventMessage) => {
      if (!agent) return;

      const action = message.userAction as A2UIUserAction | undefined;

      try {
        copilotkit.setProperties({
          ...(copilotkit.properties ?? {}),
          a2uiAction: message,
        });

        await copilotkit.runAgent({ agent });
      } finally {
        if (copilotkit.properties) {
          const { a2uiAction, ...rest } = copilotkit.properties;
          copilotkit.setProperties(rest);
        }
      }
    },
    [agent, copilotkit],
  );

  return (
    <div className="cpk:flex cpk:w-full cpk:flex-none cpk:flex-col cpk:gap-4">
      <A2UIProvider onAction={handleAction} theme={theme} catalog={catalog}>
        <SurfaceMessageProcessor
          surfaceId={surfaceId}
          operations={operations}
        />
        <A2UISurfaceOrError surfaceId={surfaceId} />
      </A2UIProvider>
    </div>
  );
}

----------------------------SurfaceMessageProcessor---------------------------------------------

function SurfaceMessageProcessor({
  surfaceId,
  operations,
}: {
  surfaceId: string;
  operations: any[];
}) {
  const { processMessages, getSurface } = useA2UIActions();
  const lastHashRef = useRef<string>("");
  useEffect(() => {
    // Skip if operations haven't actually changed (deep compare via hash).
    // ACTIVITY_DELTA + ACTIVITY_SNAPSHOT can trigger multiple renders with
    // the same logical content but different object references.
    const hash = JSON.stringify(operations);
    if (hash === lastHashRef.current) return;
    lastHashRef.current = hash;

    // Filter out createSurface if the surface already exists — the
    // MessageProcessor throws on duplicate createSurface, but content
    // snapshots always include the full operation list.
    const existing = getSurface(surfaceId);
    const ops = existing
      ? operations.filter((op) => !op?.createSurface)
      : operations;

    // Error handling is done inside A2UIProvider.processMessages
    processMessages(ops);
  }, [processMessages, getSurface, surfaceId, operations]);

  return null;
}

processMessages 通过执行不同的操作实现a2UI的渲染

processCreateSurfaceMessage(message) {
    const payload = message.createSurface;
    const { surfaceId, catalogId, theme, sendDataModel } = payload;
    // Find catalog
    const catalog = this.catalogs.find(c => c.id === catalogId);
    if (!catalog) {
        throw new A2uiStateError(`Catalog not found: ${catalogId}`);
    }
    if (this.model.getSurface(surfaceId)) {
        throw new A2uiStateError(`Surface ${surfaceId} already exists.`);
    }
    const surface = new SurfaceModel(surfaceId, catalog, theme, sendDataModel ?? false);
    this.model.addSurface(surface);
}

processUpdateComponentsMessage(message) {
    const payload = message.updateComponents;
    if (!payload.surfaceId)
        return;
    const surface = this.model.getSurface(payload.surfaceId);
    if (!surface) {
        throw new A2uiStateError(`Surface not found for message: ${payload.surfaceId}`);
    }
    for (const comp of payload.components) {
        const { id, component, ...properties } = comp;
        if (!id) {
            throw new A2uiValidationError(`Component '${component}' is missing an 'id'.`);
        }
        const existing = surface.componentsModel.get(id);
        if (existing) {
            if (component && component !== existing.type) {
                // Recreate component if type changes
                surface.componentsModel.removeComponent(id);
                const newComponent = new ComponentModel(id, component, properties);
                surface.componentsModel.addComponent(newComponent);
            }
            else {
                existing.properties = properties;
            }
        }
        else {
            if (!component) {
                throw new A2uiValidationError(`Cannot create component ${id} without a type.`);
            }
            const newComponent = new ComponentModel(id, component, properties);
            surface.componentsModel.addComponent(newComponent);
        }
    }
}

mcp 渲染逻辑

详见下面代码， 主要逻辑就是根据mcp配置通过 agent 发起请求获取到html 再通过通知方式将内容填到创建好的iframe中

/**
 * MCP Apps Extension Activity Renderer
 *
 * Renders MCP Apps UI in a sandboxed iframe with full protocol support.
 * Fetches resource content on-demand via proxied MCP requests.
 */
export const MCPAppsActivityRenderer: React.FC<MCPAppsActivityRendererProps> =
  function MCPAppsActivityRenderer({ content, agent }) {
    const containerRef = useRef<HTMLDivElement>(null);
    const iframeRef = useRef<HTMLIFrameElement | null>(null);
    const [iframeReady, setIframeReady] = useState(false);
    const [error, setError] = useState<Error | null>(null);
    const [isLoading, setIsLoading] = useState(true);
    const [iframeSize, setIframeSize] = useState<{
      width?: number;
      height?: number;
    }>({});
    const [fetchedResource, setFetchedResource] =
      useState<FetchedResource | null>(null);

    // Use refs for values that shouldn't trigger re-renders but need latest values
    const contentRef = useRef(content);
    contentRef.current = content;

    // Store agent in a ref for use in async handlers
    const agentRef = useRef(agent);
    agentRef.current = agent;

    // Ref to track fetch state - survives StrictMode remounts
    const fetchStateRef = useRef<{
      inProgress: boolean;
      promise: Promise<FetchedResource | null> | null;
      resourceUri: string | null;
    }>({ inProgress: false, promise: null, resourceUri: null });

    // Callback to send a message to the iframe
    const sendToIframe = useCallback((msg: JSONRPCMessage) => {
      if (iframeRef.current?.contentWindow) {
        console.log("[MCPAppsRenderer] Sending to iframe:", msg);
        iframeRef.current.contentWindow.postMessage(msg, "*");
      }
    }, []);

    // Callback to send a JSON-RPC response
    const sendResponse = useCallback(
      (id: string | number, result: unknown) => {
        sendToIframe({
          jsonrpc: "2.0",
          id,
          result,
        });
      },
      [sendToIframe],
    );

    // Callback to send a JSON-RPC error response
    const sendErrorResponse = useCallback(
      (id: string | number, code: number, message: string) => {
        sendToIframe({
          jsonrpc: "2.0",
          id,
          error: { code, message },
        });
      },
      [sendToIframe],
    );

    // Callback to send a notification
    const sendNotification = useCallback(
      (method: string, params?: Record<string, unknown>) => {
        sendToIframe({
          jsonrpc: "2.0",
          method,
          params: params || {},
        });
      },
      [sendToIframe],
    );

    // Effect 0: Fetch the resource content on mount
    // Uses ref-based deduplication to handle React StrictMode double-mounting
    useEffect(() => {
      const { resourceUri, serverHash, serverId } = content;

      // Check if we already have a fetch in progress for this resource
      // This handles StrictMode double-mounting - second mount reuses first mount's promise
      if (
        fetchStateRef.current.inProgress &&
        fetchStateRef.current.resourceUri === resourceUri
      ) {
        // Reuse the existing promise
        fetchStateRef.current.promise
          ?.then((resource) => {
            if (resource) {
              setFetchedResource(resource);
              setIsLoading(false);
            }
          })
          .catch((err) => {
            setError(err instanceof Error ? err : new Error(String(err)));
            setIsLoading(false);
          });
        return;
      }

      if (!agent) {
        setError(new Error("No agent available to fetch resource"));
        setIsLoading(false);
        return;
      }

      // Mark fetch as in progress
      fetchStateRef.current.inProgress = true;
      fetchStateRef.current.resourceUri = resourceUri;

      // Create the fetch promise using the queue to serialize requests
      const fetchPromise = (async (): Promise<FetchedResource | null> => {
        try {
          // Use queue to wait for agent to be idle and serialize requests
          const runResult = await mcpAppsRequestQueue.enqueue(agent, () =>
            agent.runAgent({
              forwardedProps: {
                __proxiedMCPRequest: {
                  serverHash,
                  serverId, // optional, takes precedence if provided
                  method: "resources/read",
                  params: { uri: resourceUri },
                },
              },
            }),
          );

          // Extract resource from result
          // The response format is: { contents: [{ uri, mimeType, text?, blob?, _meta? }] }
          const resultData = runResult.result as
            | { contents?: FetchedResource[] }
            | undefined;
          const resource = resultData?.contents?.[0];

          if (!resource) {
            throw new Error("No resource content in response");
          }

          return resource;
        } catch (err) {
          console.error("[MCPAppsRenderer] Failed to fetch resource:", err);
          throw err;
        } finally {
          // Mark fetch as complete
          fetchStateRef.current.inProgress = false;
        }
      })();

      // Store the promise for potential reuse
      fetchStateRef.current.promise = fetchPromise;

      // Handle the result
      fetchPromise
        .then((resource) => {
          if (resource) {
            setFetchedResource(resource);
            setIsLoading(false);
          }
        })
        .catch((err) => {
          setError(err instanceof Error ? err : new Error(String(err)));
          setIsLoading(false);
        });

      // No cleanup needed - we want the fetch to complete even if StrictMode unmounts
    }, [agent, content]);

    // Effect 1: Setup sandbox proxy iframe and communication (after resource is fetched)
    useEffect(() => {
      // Wait for resource to be fetched
      if (isLoading || !fetchedResource) {
        return;
      }

      // Capture container reference at effect start (refs are cleared during unmount)
      const container = containerRef.current;
      if (!container) {
        return;
      }

      let mounted = true;
      let messageHandler: ((event: MessageEvent) => void) | null = null;
      let initialListener: ((event: MessageEvent) => void) | null = null;
      let createdIframe: HTMLIFrameElement | null = null;

      const setup = async () => {
        try {
          // Create sandbox proxy iframe
          const iframe = document.createElement("iframe");
          createdIframe = iframe; // Track for cleanup
          iframe.style.width = "100%";
          iframe.style.height = "100px"; // Start small, will be resized by size-changed notification
          iframe.style.border = "none";
          iframe.style.backgroundColor = "transparent";
          iframe.style.display = "block";
          iframe.setAttribute(
            "sandbox",
            "allow-scripts allow-same-origin allow-forms",
          );

          // Wait for sandbox proxy to be ready
          const sandboxReady = new Promise<void>((resolve) => {
            initialListener = (event: MessageEvent) => {
              if (event.source === iframe.contentWindow) {
                if (
                  event.data?.method === "ui/notifications/sandbox-proxy-ready"
                ) {
                  if (initialListener) {
                    window.removeEventListener("message", initialListener);
                    initialListener = null;
                  }
                  resolve();
                }
              }
            };
            window.addEventListener("message", initialListener);
          });

          // Check mounted before adding to DOM (handles StrictMode double-mount)
          if (!mounted) {
            if (initialListener) {
              window.removeEventListener("message", initialListener);
              initialListener = null;
            }
            return;
          }

          // Build sandbox HTML with CSP domains from resource metadata
          const cspDomains = fetchedResource._meta?.ui?.csp?.resourceDomains;
          iframe.srcdoc = buildSandboxHTML(cspDomains);
          iframeRef.current = iframe;
          container.appendChild(iframe);

          // Wait for sandbox proxy to signal ready
          await sandboxReady;
          if (!mounted) return;

          console.log("[MCPAppsRenderer] Sandbox proxy ready");

          // Setup message handler for JSON-RPC messages from the inner iframe
          messageHandler = async (event: MessageEvent) => {
            if (event.source !== iframe.contentWindow) return;

            const msg = event.data as JSONRPCMessage;
            if (!msg || typeof msg !== "object" || msg.jsonrpc !== "2.0")
              return;

            console.log("[MCPAppsRenderer] Received from iframe:", msg);

            // Handle requests (need response)
            if (isRequest(msg)) {
              switch (msg.method) {
                case "ui/initialize": {
                  // Respond with host capabilities
                  sendResponse(msg.id, {
                    protocolVersion: PROTOCOL_VERSION,
                    hostInfo: {
                      name: "CopilotKit MCP Apps Host",
                      version: "1.0.0",
                    },
                    hostCapabilities: {
                      openLinks: {},
                      logging: {},
                    },
                    hostContext: {
                      theme: "light",
                      platform: "web",
                    },
                  });
                  break;
                }

                case "ui/message": {
                  // Add message to CopilotKit chat
                  const currentAgent = agentRef.current;

                  if (!currentAgent) {
                    console.warn(
                      "[MCPAppsRenderer] ui/message: No agent available",
                    );
                    sendResponse(msg.id, { isError: false });
                    break;
                  }

                  try {
                    const params = msg.params as {
                      role?: string;
                      content?: Array<{ type: string; text?: string }>;
                    };

                    // Extract text content from the message
                    const textContent =
                      params.content
                        ?.filter((c) => c.type === "text" && c.text)
                        .map((c) => c.text)
                        .join("\n") || "";

                    if (textContent) {
                      currentAgent.addMessage({
                        id: crypto.randomUUID(),
                        role: (params.role as "user" | "assistant") || "user",
                        content: textContent,
                      });
                    }
                    sendResponse(msg.id, { isError: false });
                  } catch (err) {
                    console.error("[MCPAppsRenderer] ui/message error:", err);
                    sendResponse(msg.id, { isError: true });
                  }
                  break;
                }

                case "ui/open-link": {
                  // Open URL in new tab
                  const url = msg.params?.url as string | undefined;
                  if (url) {
                    window.open(url, "_blank", "noopener,noreferrer");
                    sendResponse(msg.id, { isError: false });
                  } else {
                    sendErrorResponse(msg.id, -32602, "Missing url parameter");
                  }
                  break;
                }

                case "tools/call": {
                  // Proxy tool call to MCP server via agent.runAgent()
                  const { serverHash, serverId } = contentRef.current;
                  const currentAgent = agentRef.current;

                  if (!serverHash) {
                    sendErrorResponse(
                      msg.id,
                      -32603,
                      "No server hash available for proxying",
                    );
                    break;
                  }

                  if (!currentAgent) {
                    sendErrorResponse(
                      msg.id,
                      -32603,
                      "No agent available for proxying",
                    );
                    break;
                  }

                  try {
                    // Use queue to wait for agent to be idle and serialize requests
                    const runResult = await mcpAppsRequestQueue.enqueue(
                      currentAgent,
                      () =>
                        currentAgent.runAgent({
                          forwardedProps: {
                            __proxiedMCPRequest: {
                              serverHash,
                              serverId, // optional, takes precedence if provided
                              method: "tools/call",
                              params: msg.params,
                            },
                          },
                        }),
                    );

                    // The result from runAgent contains the MCP response
                    sendResponse(msg.id, runResult.result || {});
                  } catch (err) {
                    console.error("[MCPAppsRenderer] tools/call error:", err);
                    sendErrorResponse(msg.id, -32603, String(err));
                  }
                  break;
                }

                default:
                  sendErrorResponse(
                    msg.id,
                    -32601,
                    `Method not found: ${msg.method}`,
                  );
              }
            }

            // Handle notifications (no response needed)
            if (isNotification(msg)) {
              switch (msg.method) {
                case "ui/notifications/initialized": {
                  console.log("[MCPAppsRenderer] Inner iframe initialized");
                  if (mounted) {
                    setIframeReady(true);
                  }
                  break;
                }

                case "ui/notifications/size-changed": {
                  const { width, height } = msg.params || {};
                  console.log("[MCPAppsRenderer] Size change:", {
                    width,
                    height,
                  });
                  if (mounted) {
                    setIframeSize({
                      width: typeof width === "number" ? width : undefined,
                      height: typeof height === "number" ? height : undefined,
                    });
                  }
                  break;
                }

                case "notifications/message": {
                  // Logging notification from the app
                  console.log("[MCPAppsRenderer] App log:", msg.params);
                  break;
                }
              }
            }
          };

          window.addEventListener("message", messageHandler);

          // Extract HTML content from fetched resource
          let html: string;
          if (fetchedResource.text) {
            html = fetchedResource.text;
          } else if (fetchedResource.blob) {
            html = atob(fetchedResource.blob);
          } else {
            throw new Error("Resource has no text or blob content");
          }

          // Send the resource content to the sandbox proxy
          sendNotification("ui/notifications/sandbox-resource-ready", { html });
        } catch (err) {
          console.error("[MCPAppsRenderer] Setup error:", err);
          if (mounted) {
            setError(err instanceof Error ? err : new Error(String(err)));
          }
        }
      };

      setup();

      return () => {
        mounted = false;
        // Clean up initial listener if still active
        if (initialListener) {
          window.removeEventListener("message", initialListener);
          initialListener = null;
        }
        if (messageHandler) {
          window.removeEventListener("message", messageHandler);
        }
        // Remove the iframe we created (using tracked reference, not DOM query)
        // This works even if containerRef.current is null during unmount
        if (createdIframe) {
          createdIframe.remove();
          createdIframe = null;
        }
        iframeRef.current = null;
      };
    }, [
      isLoading,
      fetchedResource,
      sendNotification,
      sendResponse,
      sendErrorResponse,
    ]);

    // Effect 2: Update iframe size when it changes
    useEffect(() => {
      if (iframeRef.current) {
        if (iframeSize.width !== undefined) {
          // Use minWidth with min() to allow expansion but cap at 100%
          iframeRef.current.style.minWidth = `min(${iframeSize.width}px, 100%)`;
          iframeRef.current.style.width = "100%";
        }
        if (iframeSize.height !== undefined) {
          iframeRef.current.style.height = `${iframeSize.height}px`;
        }
      }
    }, [iframeSize]);

    // Effect 3: Send tool input when iframe ready
    useEffect(() => {
      if (iframeReady && content.toolInput) {
        console.log("[MCPAppsRenderer] Sending tool input:", content.toolInput);
        sendNotification("ui/notifications/tool-input", {
          arguments: content.toolInput,
        });
      }
    }, [iframeReady, content.toolInput, sendNotification]);

    // Effect 4: Send tool result when iframe ready
    useEffect(() => {
      if (iframeReady && content.result) {
        console.log("[MCPAppsRenderer] Sending tool result:", content.result);
        sendNotification("ui/notifications/tool-result", content.result);
      }
    }, [iframeReady, content.result, sendNotification]);

    // Determine border styling based on prefersBorder metadata from fetched resource
    // true = show border/background, false = none, undefined = host decides (we default to none)
    const prefersBorder = fetchedResource?._meta?.ui?.prefersBorder;
    const borderStyle =
      prefersBorder === true
        ? {
            borderRadius: "8px",
            backgroundColor: "#f9f9f9",
            border: "1px solid #e0e0e0",
          }
        : {};

    return (
      <div
        ref={containerRef}
        style={{
          width: "100%",
          height: iframeSize.height ? `${iframeSize.height}px` : "auto",
          minHeight: "100px",
          overflow: "hidden",
          position: "relative",
          ...borderStyle,
        }}
      >
        {isLoading && (
          <div style={{ padding: "1rem", color: "#666" }}>Loading...</div>
        )}
        {error && (
          <div style={{ color: "red", padding: "1rem" }}>
            Error: {error.message}
          </div>
        )}
      </div>
    );
  };

open Generative UI 渲染逻辑

全开放式渲染逻辑过程直接从后端agent content 中获取html 再通过构建沙河环境，将html 渲染到iframe中

function OpenGenerativeUIActivityRendererInner({ content }: InnerProps) {
   const initialHeight = content.initialHeight ?? 200;
   const [autoHeight, setAutoHeight] = useState<number | null>(null);
   const sandboxFunctions = useSandboxFunctions();

   const localApi = useMemo(() => {
     const api: Record<string, Function> = {};
     for (const fn of sandboxFunctions) {
       api[fn.name] = fn.handler;
     }
     return api;
   }, [sandboxFunctions]);

   // Join html chunks only when streaming is complete
   const fullHtml =
     content.htmlComplete && content.html?.length
       ? content.html.join("")
       : undefined;
   // CSS from the dedicated parameter (available once cssComplete)
   const css = content.cssComplete ? content.css : undefined;

   ....
}

CopilotRuntime

本质是一个代理, 或者代理适配器(多个agent存在时)，寻找后端agent 服务，并将前端请求转发给后端agent 服务

On the server, CopilotRuntime accepts a map of AG-UI AbstractAgent instances. A framework adapter, an HttpAgent pointing at a remote server, and a custom implementation all use the same request path:

• The runtime resolves the target agent by ID.
• It clones the agent for request isolation and supplies messages, state, and thread context.
• AgentRunner executes the agent and receives AG-UI events.
• The runtime encodes those events as SSE and streams them to the frontend proxy.
• The backend framework can change without forcing a corresponding change to the frontend AG-UI contract.

官方文档

它是一个框架无关代理，所以也可以用在支持Fetch API的node 层 运行时中
Deploy to any runtime

AGUI 协议

abstractAgent

[AbstractAgent Api] (https://docs.ag-ui.com/sdk/js/client/abstract-agent)
AbstractAgent 源码

abstract run(input: RunAgentInput): Observable<BaseEvent>;

public async runAgent(
    parameters?: RunAgentParameters,
    subscriber?: AgentSubscriber,
  ): Promise<RunAgentResult> {
    try {
      this.isRunning = true;
      this.agentId = this.agentId ?? uuidv4();
      const input = this.prepareRunAgentInput(parameters);

      this.debugLogger?.lifecycle("LIFECYCLE", "Run started:", {
        agentId: this.agentId,
        threadId: this.threadId,
      });

      let result: any = undefined;
      const currentMessageIds = new Set(this.messages.map((message) => message.id));

      const subscribers: AgentSubscriber[] = [
        {
          onRunFinishedEvent: (params) => {
            if (params.outcome === "success") {
              result = params.result;
            }
          },
        },
        ...this.subscribers,
        subscriber ?? {},
      ];

      await this.onInitialize(input, subscribers);

      // Per-run detachment signal + completion promise
      this.activeRunDetach$ = new Subject<void>();
      let resolveActiveRunCompletion: (() => void) | undefined;
      this.activeRunCompletionPromise = new Promise<void>((resolve) => {
        resolveActiveRunCompletion = resolve;
      });

      // 没有中间件的话直接运行run 有种中间件的话，会先运行中间件，初始值是agent本身，reduceRight 到最后是会执行agent.run(input) 方法

      // pipe: rxjs api https://rxjs.dev/api/index/function/pipe

      const pipeline = pipe(
        () => {
          // Build middleware chain using reduceRight so middlewares can intercept runs.
          if (this.middlewares.length === 0) {
            return this.run(input);
          }

          const chainedAgent = this.middlewares.reduceRight(
            (nextAgent: AbstractAgent, middleware) =>
              ({
                run: (i: RunAgentInput) => middleware.run(i, nextAgent),
                get messages() {
                  return nextAgent.messages;
                },
                get state() {
                  return nextAgent.state;
                },
              }) as AbstractAgent,
            this, // Original agent is the final 'next'
          );

          return chainedAgent.run(input);
        },
        transformChunks(this.debugLogger),
        verifyEvents(this.debugLogger),
        // Stop processing immediately when this run is detached
        (source$) => source$.pipe(takeUntil(this.activeRunDetach$!)),
        (source$) => this.apply(input, source$, subscribers),
        (source$) => this.processApplyEvents(input, source$, subscribers),
        catchError((error) => {
          this.debugLogger?.lifecycle("LIFECYCLE", "Run errored:", {
            agentId: this.agentId,
            error: error instanceof Error ? error.message : String(error),
          });
          this.isRunning = false;
          return this.onError(input, error, subscribers);
        }),
        finalize(() => {
          this.debugLogger?.lifecycle("LIFECYCLE", "Run finished:", {
            agentId: this.agentId,
            threadId: this.threadId,
          });
          this.isRunning = false;
          void this.onFinalize(input, subscribers);
          resolveActiveRunCompletion?.();
          resolveActiveRunCompletion = undefined;
          this.activeRunCompletionPromise = undefined;
          this.activeRunDetach$ = undefined;
        }),
      );

      await lastValueFrom(pipeline(of(null)));
      const newMessages = structuredClone_(this.messages).filter(
        (message: Message) => !currentMessageIds.has(message.id),
      );
      return { result, newMessages };
    } finally {
      this.isRunning = false;
    }
  }

httpAgent

[HttpAgent Api] (https://docs.ag-ui.com/sdk/js/client/http-agent)

httpAgent 基于 abstractAgent 抽象类

HttpAgent 源码

httpAgent 会实现 run 函数，发起真正的http 请求

/**
   * Returns the fetch config for the http request.
   * Override this to customize the request.
   *
   * @returns The fetch config for the http request.
   */
  protected requestInit(input: RunAgentInput): RequestInit {
    return {
      method: "POST",
      headers: {
        ...this.headers,
        "Content-Type": "application/json",
        Accept: "text/event-stream",
      },
      body: JSON.stringify(input),
      signal: this.abortController.signal,
    };
  }
  run(input: RunAgentInput): Observable<BaseEvent> {
    const httpEvents = runHttpRequest(() => this.fetch(this.url, this.requestInit(input)));
    return transformHttpEventStream(httpEvents, this.debugLogger);
  }

runHttpRequest

runHttpRequest 处理流数据 转成 HttpEventType 流

export enum HttpEventType {
  HEADERS = "headers",
  DATA = "data",
}

export const runHttpRequest = (
  fetchResponse: () => Promise<Response>,
): Observable<HttpEvent> => {
  // Defer the fetch so that it's executed when subscribed to
  return defer(() => from(fetchResponse())).pipe(
    switchMap((response) => {
      if (!response.ok) {
        const contentType = response.headers.get("content-type") || "";
        // Read the (small) error body once, then error the stream
        return from(response.text()).pipe(
          mergeMap((text) => {
            let payload: unknown = text;
            if (contentType.includes("application/json")) {
              try { payload = JSON.parse(text); } catch {/* keep raw text */}
            }
            const err: any = new Error(
              `HTTP ${response.status}: ${typeof payload === "string" ? payload : JSON.stringify(payload)}`
            );
            err.status = response.status;
            err.payload = payload;
            return throwError(() => err);
          })
        );
      }
      // Emit headers event first
      const headersEvent: HttpHeadersEvent = {
        type: HttpEventType.HEADERS,
        status: response.status,
        headers: response.headers,
      };

      const reader = response.body?.getReader();
      if (!reader) {
        return throwError(() => new Error("Failed to getReader() from response"));
      }

      return new Observable<HttpEvent>((subscriber) => {
        // Emit headers event first
        subscriber.next(headersEvent);

        (async () => {
          try {
            while (true) {
              const { done, value } = await reader.read();
              if (done) break;
              // Emit data event instead of raw Uint8Array
              const dataEvent: HttpDataEvent = {
                type: HttpEventType.DATA,
                data: value,
              };
              subscriber.next(dataEvent);
            }
            subscriber.complete();
          } catch (error) {
            subscriber.error(error);
          }
        })();

        return () => {
          reader.cancel().catch((error) => {
            if ((error as DOMException)?.name === "AbortError") {
              return;
            }

            throw error;
          });
        };
      });
    }),
  );
};

transformHttpEventStream

export enum EventType {
  TEXT_MESSAGE_START = "TEXT_MESSAGE_START",
  TEXT_MESSAGE_CONTENT = "TEXT_MESSAGE_CONTENT",
  TEXT_MESSAGE_END = "TEXT_MESSAGE_END",
  TEXT_MESSAGE_CHUNK = "TEXT_MESSAGE_CHUNK",
  TOOL_CALL_START = "TOOL_CALL_START",
  TOOL_CALL_ARGS = "TOOL_CALL_ARGS",
  TOOL_CALL_END = "TOOL_CALL_END",
  TOOL_CALL_CHUNK = "TOOL_CALL_CHUNK",
  TOOL_CALL_RESULT = "TOOL_CALL_RESULT",
  THINKING_TEXT_MESSAGE_END = "THINKING_TEXT_MESSAGE_END",
  STATE_SNAPSHOT = "STATE_SNAPSHOT",
  STATE_DELTA = "STATE_DELTA",
  MESSAGES_SNAPSHOT = "MESSAGES_SNAPSHOT",
  ACTIVITY_SNAPSHOT = "ACTIVITY_SNAPSHOT",
  ACTIVITY_DELTA = "ACTIVITY_DELTA",
  RAW = "RAW",
  CUSTOM = "CUSTOM",
  RUN_STARTED = "RUN_STARTED",
  RUN_FINISHED = "RUN_FINISHED",
  RUN_ERROR = "RUN_ERROR",
  STEP_STARTED = "STEP_STARTED",
  STEP_FINISHED = "STEP_FINISHED",
  REASONING_START = "REASONING_START",
  REASONING_MESSAGE_START = "REASONING_MESSAGE_START",
  REASONING_MESSAGE_CONTENT = "REASONING_MESSAGE_CONTENT",
  REASONING_MESSAGE_END = "REASONING_MESSAGE_END",
  REASONING_MESSAGE_CHUNK = "REASONING_MESSAGE_CHUNK",
  REASONING_END = "REASONING_END",
  REASONING_ENCRYPTED_VALUE = "REASONING_ENCRYPTED_VALUE",
}

/**
 * Transforms HTTP events into BaseEvents using the appropriate format parser based on content type.
 */

export const transformHttpEventStream = (
  source$: Observable<HttpEvent>,
  debugLogger?: DebugLoggerInput,
): Observable<BaseEvent> => {

  .....

   // If we get headers and haven't initialized a parser yet, check content type
      if (event.type === HttpEventType.HEADERS && !parserInitialized) {
        parserInitialized = true;
        const contentType = event.headers.get("content-type");

        log?.lifecycle("HTTP", "Stream format detected:", {
          contentType,
          parser: contentType === proto.AGUI_MEDIA_TYPE ? "protobuf" : "sse",
        });

        // Choose parser based on content type
        if (contentType === proto.AGUI_MEDIA_TYPE) {
          // Use protocol buffer parser
          parseProtoStream(bufferSubject).subscribe({
            next: (event) => eventSubject.next(event),
            error: (err) => eventSubject.error(err),
            complete: () => eventSubject.complete(),
          });
        } else {
          // Use SSE JSON parser for all other cases
          parseSSEStream(bufferSubject, log).subscribe({
            next: (json) => {
              try {
                const parsedEvent = EventSchemas.parse(json);
                log?.event("HTTP", "Event validated:", parsedEvent, {
                  type: parsedEvent.type,
                  valid: true,
                });
                eventSubject.next(parsedEvent as BaseEvent);
              } catch (err) {
                log?.event("HTTP", "Event invalid:", { json, error: String(err) });
                eventSubject.error(err);
              }
            },
            error: (err) => {
              ....
              return eventSubject.error(err);
            },
            complete: () => eventSubject.complete(),
          });
        }
      } else if (!parserInitialized) {
        eventSubject.error(new Error("No headers event received before data events"));
      }

}

parseSSEStream

源码

/**
 * Parses a stream of HTTP events into a stream of JSON objects using Server-Sent Events (SSE) format.
 * Strictly follows the SSE standard where:
 * - Events are separated by double newlines ('\n\n')
 * - Only 'data:' prefixed lines are processed
 * - Multi-line data events are supported and joined
 * - Non-data fields (event, id, retry) are ignored
 */

 next: (event: HttpEvent) => {
      if (event.type === HttpEventType.HEADERS) {
        return;
      }

      if (event.type === HttpEventType.DATA && event.data) {
        // Decode chunk carefully to handle UTF-8
        const text = decoder.decode(event.data, { stream: true });
        buffer += text;

        // Process complete events (separated by double newlines)
        const events = buffer.split(/\n\n/);
        // Keep the last potentially incomplete event in buffer
        buffer = events.pop() || "";

        for (const event of events) {
          processSSEEvent(event);
        }
      }
    },

/**
   * Helper function to process an SSE event.
   * Extracts and joins data lines, then parses the result as JSON.
   *
   * Follows the SSE spec by processing lines starting with 'data:',
   * ignoring a single space if it is present after the colon.
   *
   * @param eventText The raw event text to process
   */
  function processSSEEvent(eventText: string) {
    const lines = eventText.split("\n");
    const dataLines: string[] = [];

    for (const line of lines) {
      if (line.startsWith("data:")) {
        // Remove 'data:' prefix, and optionally a single space afterwards
        dataLines.push(line.slice(5).replace(/^ /, ""));
      }
    }

    // Only process if we have data lines
    if (dataLines.length > 0) {
      try {
        // Join multi-line data and parse JSON
        const jsonStr = dataLines.join("\n");
        const json = JSON.parse(jsonStr);
        log?.event("SSE", "Event received:", json, { type: json.type });
        jsonSubject.next(json);
      } catch (err) {
        jsonSubject.error(err);
      }
    }
  }

parseProtoStream

源码

/**
 * Parses a stream of HTTP events into a stream of BaseEvent objects using Protocol Buffer format.
 * Each message is prefixed with a 4-byte length header (uint32 in big-endian format)
 * followed by the protocol buffer encoded message.
 */

   next: (event: HttpEvent) => {
      if (event.type === HttpEventType.HEADERS) {
        return;
      }

      if (event.type === HttpEventType.DATA && event.data) {
        // Append the new data to our buffer
        const newBuffer = new Uint8Array(buffer.length + event.data.length);
        newBuffer.set(buffer, 0);
        newBuffer.set(event.data, buffer.length);
        buffer = newBuffer;

        // Process as many complete messages as possible
        processBuffer();
      }
    },

 /**
   * Process as many complete messages as possible from the buffer
   */
function processBuffer() {
    // Keep processing while we have enough data for at least a header (4 bytes)
    while (buffer.length >= 4) {
      // Read message length from the first 4 bytes (big-endian uint32)
      const view = new DataView(buffer.buffer, buffer.byteOffset, 4);
      const messageLength = view.getUint32(0, false); // false = big-endian

      // Check if we have the complete message (header + message body)
      const totalLength = 4 + messageLength;
      if (buffer.length < totalLength) {
        // Not enough data yet, wait for more
        break;
      }

      try {
        // Extract the message (skipping the 4-byte header)
        const message = buffer.slice(4, totalLength);

        // Decode the protocol buffer message using the imported decode function
        const event = proto.decode(message);

        // Emit the parsed event
        eventSubject.next(event);

        // Remove the processed message from the buffer
        buffer = buffer.slice(totalLength);
      } catch (error: unknown) {
        const errorMessage = error instanceof Error ? error.message : String(error);
        eventSubject.error(new Error(`Failed to decode protocol buffer message: ${errorMessage}`));
        return;
      }
    }
  }

为什么默认 copilotKit 实例不同动作 可以发送不同http path 请求

从官网文档中可以看出，默认的 copilotKit 实例不同动作会基于 basePath 发送不同的http path 请求
实际请求时确实从寻找agent, 链接agent 服务，到向agent 提问 都发送不同的http path 请求

原因：
useAgent() 函数会返回一个 ProxiedCopilotRuntimeAgent 对象
ProxiedCopilotRuntimeAgent 从httpAgent继承，
ProxiedCopilotRuntimeAgent 这个类会重新生成run 的请求url, 重写 connect 方法过程会指定，

/**
 * ProxiedCopilotRuntimeAgent 从httpAgent继承
 * 
 */


super({
  ...config,
  url: runUrl,
});

/**   run  ********************/

public run(input: RunAgentInput): Observable<BaseEvent> {
    if (
      this.runtimeMode === "pending" ||
      (this.transport === "auto" &&
        this.runtimeMode !== RUNTIME_MODE_INTELLIGENCE)
    ) {
      return defer(() => from(this.ensureRuntimeConfiguration())).pipe(
        switchMap(() => this.run(input)),
      );
    }
    if (this.runtimeMode === RUNTIME_MODE_INTELLIGENCE) {
      return this.#runViaDelegate(input);
    }
    return this.#runViaHttp(input);
  }

 #runViaDelegate(input: RunAgentInput): Observable<BaseEvent> {
    return defer(() => from(this.resolveDelegate())).pipe(
      switchMap((delegate) => withAbortErrorHandling(delegate.run(input))),
    );
  }

  #runViaHttp(input: RunAgentInput): Observable<BaseEvent> {
    if (this.transport === "single") {
      if (!this.singleEndpointUrl) {
        throw new Error("Single endpoint transport requires a runtimeUrl");
      }

      const requestInit = this.createSingleRouteRequestInit(
        input,
        "agent/run",
        {
          agentId: this.routedAgentId(),
        },
      );
      const httpEvents = runHttpRequest(() =>
        this.fetch(this.singleEndpointUrl!, requestInit),
      );
      return withAbortErrorHandling(transformHttpEventStream(httpEvents));
    }
    /**
     * POST /api/copilotkit/agent/:agentId/run
     */
    return withAbortErrorHandling(super.run(input));
  }


/**
 * 
 * /api/copilotkit/info 
 */


private async fetchRuntimeInfo(): Promise<RuntimeInfo> {
    const headers: Record<string, string> = {
      ...this.headers,
    };

    if (this.transport === "auto") {
      return this.fetchRuntimeInfoAutoDetect(headers);
    }

    let init: RequestInit;
    let url: string;

    if (this.transport === "single") {
      if (!this.singleEndpointUrl) {
        throw new Error("Single endpoint transport requires a runtimeUrl");
      }
      if (!headers["Content-Type"]) {
        headers["Content-Type"] = "application/json";
      }
      url = this.runtimeUrl!;
      init = { method: "POST", body: JSON.stringify({ method: "info" }) };
    } else {
      url = `${this.runtimeUrl}/info`;
      init = {};
    }
    ....

}

ProxiedCopilotRuntimeAgent 源码