Add CrewAI support (CopilotKit#1308)

Co-authored-by: Suhas Deshpande <suhasdeshpande@users.noreply.github.com>
Co-authored-by: Ariel Weinberger <Weinberger.Ariel@gmail.com>
Co-authored-by: Suhas Deshpande <suhas2u@gmail.com>

Author: 3 people

CopilotKit/.changeset/khaki-spoons-relate.md

@@ -0,0 +1,7 @@
+---
+"@copilotkit/runtime": patch
+---
+
+- Added RemoteAgentAdapter and implemented CopilotKit (protocol & events).
+- Integrated CrewAI (with a prototype, HITL, event system, and chat/demo tweaks).
+- Updated docs and cleaned up code (fixing stdout issues, restricting Python version, and streamlining demos).

CopilotKit/.changeset/slow-buckets-own.md

@@ -0,0 +1,5 @@
+---
+"@copilotkit/runtime": minor
+---
+
+- CrewAI support

CopilotKit/.vscode/cspell.json

@@ -17,6 +17,7 @@
     "pino",
     "langsmith",
     "readables",
-    "uuidv4"
+    "uuidv4",
+    "crewai"
   ]
 }

CopilotKit/packages/runtime/src/graphql/resolvers/copilot.resolver.ts

@@ -196,6 +196,10 @@ export class CopilotResolver {
       rejectOutputMessagesPromise = reject;
     });
 
+    if (copilotCloudPublicApiKey) {
+      ctx.properties["copilotCloudPublicApiKey"] = copilotCloudPublicApiKey;
+    }
+
     logger.debug("Processing");
     const {
       eventSource,

CopilotKit/packages/runtime/src/lib/runtime/__tests__/remote-action-constructors.test.ts

@@ -1,3 +1,4 @@
+import "reflect-metadata";
 import { TextEncoder } from "util";
 import { RemoteLangGraphEventSource } from "../../../agents/langgraph/event-source";
 import telemetry from "../../telemetry-client";
@@ -7,6 +8,7 @@ import {
   createHeaders,
 } from "../remote-action-constructors";
 import { execute } from "../remote-lg-action";
+import { ReplaySubject } from "rxjs";
 
 // Mock external dependencies
 jest.mock("../remote-lg-action", () => ({
@@ -49,7 +51,7 @@ beforeEach(() => {
 
 describe("remote action constructors", () => {
   describe("constructLGCRemoteAction", () => {
-    it("should create an agent with langGraphAgentHandler that processes events", async () => {
+    it("should create an agent with remoteAgentHandler that processes events", async () => {
       // Arrange: simulate execute returning a dummy ReadableStream
       const dummyEncodedEvent = new TextEncoder().encode(JSON.stringify({ event: "test" }) + "\n");
       const readerMock = {
@@ -72,7 +74,7 @@ describe("remote action constructors", () => {
         processLangGraphEvents: processLangGraphEventsMock,
       }));
 
-      // Act: build the action and call langGraphAgentHandler
+      // Act: build the action and call remoteAgentHandler
       const actions = constructLGCRemoteAction({
         endpoint,
         graphqlContext,
@@ -84,7 +86,7 @@ describe("remote action constructors", () => {
       const action = actions[0];
       expect(action.name).toEqual(dummyAgent.name);
 
-      const result = await action.langGraphAgentHandler({
+      const result = await action.remoteAgentHandler({
         name: dummyAgent.name,
         actionInputsWithoutAgents: [],
         threadId: "thread1",
@@ -160,26 +162,46 @@ describe("remote action constructors", () => {
     });
 
     it("should create remote agent handler that processes events", async () => {
-      // Arrange: mock fetch for agent handler to return a dummy stream
-      const dummyEncodedAgentEvent = new TextEncoder().encode('{"event":"data"}\n');
-      const agentReaderMock = {
-        read: jest
-          .fn()
-          .mockResolvedValueOnce({ done: false, value: dummyEncodedAgentEvent })
-          .mockResolvedValueOnce({ done: true, value: new Uint8Array() }),
-      };
-      const dummyStreamResponse = {
-        getReader: () => agentReaderMock,
+      const json = {
+        agents: [
+          {
+            name: "agent2",
+            description: "agent desc",
+            type: "langgraph", // Add type to match RemoteAgentType.LangGraph
+          },
+        ],
+        actions: [
+          {
+            name: "action1",
+            description: "action desc",
+            parameters: { param: "value" },
+          },
+        ],
       };
-      global.fetch = jest.fn().mockResolvedValue({
-        ok: true,
-        text: jest.fn().mockResolvedValue("ok"),
-        body: dummyStreamResponse,
-      });
 
-      const processLangGraphEventsMock = jest.fn(() => "agent events processed");
+      const dummyEncodedAgentEvent = new TextEncoder().encode('{"type":"data","content":"test"}\n');
+
+      const mockResponse = new Response(
+        new ReadableStream({
+          start(controller) {
+            controller.enqueue(dummyEncodedAgentEvent);
+            controller.close();
+          },
+        }),
+        {
+          status: 200,
+          statusText: "OK",
+          headers: new Headers({
+            "content-type": "application/json",
+          }),
+        },
+      );
+
+      global.fetch = jest.fn().mockResolvedValue(mockResponse);
+
+      const processLangGraphEventsMock = jest.fn().mockResolvedValue("agent events processed");
       (RemoteLangGraphEventSource as jest.Mock).mockImplementation(() => ({
-        eventStream$: { next: jest.fn(), complete: jest.fn(), error: jest.fn() },
+        eventStream$: new ReplaySubject(),
         processLangGraphEvents: processLangGraphEventsMock,
       }));
 
@@ -192,29 +214,17 @@ describe("remote action constructors", () => {
         messages: [],
         agentStates,
       });
-      // The remote agent is the second item in the array
-      expect(actionsArray).toHaveLength(2);
-      const remoteAgentHandler = (actionsArray[1] as any).langGraphAgentHandler;
+
+      const remoteAgentHandler = (actionsArray[1] as any).remoteAgentHandler;
       const result = await remoteAgentHandler({
         name: "agent2",
         actionInputsWithoutAgents: [],
         threadId: "thread2",
         nodeName: "node2",
-        additionalMessages: [],
-        metaEvents: [],
       });
+
       expect(processLangGraphEventsMock).toHaveBeenCalled();
       expect(result).toBe("agent events processed");
-
-      // Check telemetry.capture for agent execution
-      expect(telemetry.capture).toHaveBeenCalledWith(
-        "oss.runtime.remote_action_executed",
-        expect.objectContaining({
-          agentExecution: true,
-          type: "self-hosted",
-          agentsAmount: 1,
-        }),
-      );
     });
   });
 

CopilotKit/packages/runtime/src/lib/runtime/copilot-runtime.ts

@@ -39,8 +39,8 @@ import { Message } from "../../graphql/types/converted";
 import { ForwardedParametersInput } from "../../graphql/inputs/forwarded-parameters.input";
 
 import {
-  isLangGraphAgentAction,
-  LangGraphAgentAction,
+  isRemoteAgentAction,
+  RemoteAgentAction,
   EndpointType,
   setupRemoteActions,
   EndpointDefinition,
@@ -305,7 +305,7 @@ please use an LLM adapter instead.`,
           (action) =>
             // TODO-AGENTS: do not exclude ALL server side actions
             !serverSideActions.find((serverSideAction) => serverSideAction.name == action.name),
-          // !isLangGraphAgentAction(
+          // !isRemoteAgentAction(
           //   serverSideActions.find((serverSideAction) => serverSideAction.name == action.name),
           // ),
         ),
@@ -322,7 +322,6 @@ please use an LLM adapter instead.`,
   }
 
   async discoverAgentsFromEndpoints(graphqlContext: GraphQLContext): Promise<AgentWithEndpoint[]> {
-    const headers = createHeaders(null, graphqlContext);
     const agents = this.remoteEndpointDefinitions.reduce(
       async (acc: Promise<Agent[]>, endpoint) => {
         const agents = await acc;
@@ -355,12 +354,12 @@ please use an LLM adapter instead.`,
             description: string;
           }>;
         }
-
-        const fetchUrl = `${(endpoint as CopilotKitEndpoint).url}/info`;
+        const cpkEndpoint = endpoint as CopilotKitEndpoint;
+        const fetchUrl = `${endpoint.url}/info`;
         try {
           const response = await fetch(fetchUrl, {
             method: "POST",
-            headers,
+            headers: createHeaders(cpkEndpoint.onBeforeRequest, graphqlContext),
             body: JSON.stringify({ properties: graphqlContext.properties }),
           });
           if (!response.ok) {
@@ -444,11 +443,12 @@ please use an LLM adapter instead.`,
       agentWithEndpoint.endpoint.type === EndpointType.CopilotKit ||
       !("type" in agentWithEndpoint.endpoint)
     ) {
-      const fetchUrl = `${(agentWithEndpoint.endpoint as CopilotKitEndpoint).url}/agents/state`;
+      const cpkEndpoint = agentWithEndpoint.endpoint as CopilotKitEndpoint;
+      const fetchUrl = `${cpkEndpoint.url}/agents/state`;
       try {
         const response = await fetch(fetchUrl, {
           method: "POST",
-          headers,
+          headers: createHeaders(cpkEndpoint.onBeforeRequest, graphqlContext),
           body: JSON.stringify({
             properties: graphqlContext.properties,
             threadId,
@@ -505,8 +505,8 @@ please use an LLM adapter instead.`,
     const messages = convertGqlInputToMessages(rawMessages);
 
     const currentAgent = serverSideActions.find(
-      (action) => action.name === agentName && isLangGraphAgentAction(action),
-    ) as LangGraphAgentAction;
+      (action) => action.name === agentName && isRemoteAgentAction(action),
+    ) as RemoteAgentAction;
 
     if (!currentAgent) {
       throw new CopilotKitAgentDiscoveryError({ agentName });
@@ -519,9 +519,9 @@ please use an LLM adapter instead.`,
       .filter(
         (action) =>
           // Case 1: Keep all regular (non-agent) actions
-          !isLangGraphAgentAction(action) ||
+          !isRemoteAgentAction(action) ||
           // Case 2: For agent actions, keep all except self (prevent infinite loops)
-          (isLangGraphAgentAction(action) && action.name !== agentName) /* prevent self-calls */,
+          (isRemoteAgentAction(action) && action.name !== agentName) /* prevent self-calls */,
       )
       .map((action) => ({
         name: action.name,
@@ -542,7 +542,7 @@ please use an LLM adapter instead.`,
     });
     try {
       const eventSource = new RuntimeEventSource();
-      const stream = await currentAgent.langGraphAgentHandler({
+      const stream = await currentAgent.remoteAgentHandler({
         name: agentName,
         threadId,
         nodeName,