Extensions

Extensions are at the core of Kairo's power. They allow you to add new capabilities to the language model by integrating external tools, data sources, or specialized logic.

What is an Extension?

An extension is a decoupled module that communicates with the LmPipeline using the Kairo Extension Protocol. This protocol is based on JSON-RPC and allows extensions to run in different environments (e.g., local process, web worker, remote server).

Available Extensions

Implemented Extensions

MCP Extension (extensions/mcp)

The Model Context Protocol (MCP) is a standard for connecting AI models to data and tools. The Kairo MCP extension allows you to use any MCP-compliant server as a source of tools and context for your Kairo applications.

RAG Local (extensions/rag-local)

A built-in implementation of Retrieval-Augmented Generation (RAG) that runs locally. It handles document indexing and similarity search to provide the model with relevant context from your own files.

Planned Extensions (Contributions Welcome)

The following extensions have had their corresponding folders created in the extensions/ directory and are planned for future development. We welcome community contributions to help bring these to life:

• Computer Controller (extensions/computer-controller): An extension that allows the model to interact with the host operating system, performing tasks like file system access or system monitoring (use with caution!)
• Code Execution (extensions/code-execution): Provides sandboxed execution environments for running code generated by the model
• Context Storage (extensions/context-storage): Offers adapters for persisting Kairo's pipeline context to external databases or file systems
• Memory (extensions/memory): Implements long-term memory retrieval and management for agents
• RAG Remote (extensions/rag): Remote Retrieval-Augmented Generation capabilities interacting with external vector databases
• Skills (extensions/skills): Support for the Agent Skills specification, providing standardized, task-specific capabilities for AI agents

Implementation Mechanism

Kairo extensions follow a Server-Client model centered around the ExtensionProtocol.

1. ExtensionServer: Your extension logic lives here. It registers request handlers for specific methods (e.g., tool-injection or custom methods).
2. Transport: The server is bound to a Transport (Stdio, HTTP, or Worker), which handles the raw bytes/messages over the wire.
3. LmPipelineClient: On the application side, the LmPipeline uses an ExtensionClient to communicate with your server through a matching transport client.

Building Your Own Extension

To create a new extension, you don't need to rebuild the Kairo core. You just need to implement the server side.

Step 1: Define Your Tools (Zod)

import { z } from "zod";

const SearchToolSchema = z.object({
  query: z.string().describe("The search query"),
});

Step 2: Implement the Extension Server

import { ExtensionServer, LMPipelineExtension } from "@kairo/core";
import { StdioTransportServer } from "@kairo/extension-transport-stdio";

const server = new ExtensionServer({
  id: "my-search-extension",
  name: "Searcher",
  version: "1.0.0",
});

// Use LMPipelineExtension builder to implement tool injection capability
LMPipelineExtension.init(server, { supportsToolInjection: true }).build(() => {
  // Use a map to track long-running tasks
  const tasks = new Map<string, any>();

  return {
    injectTools: async () => ({
      result: {
        tools: [
          {
            type: "function",
            name: "web_search",
            description: "Search the web for info",
            parameters: SearchToolSchema,
          },
        ],
      },
    }),
    executeTool: async (input) => {
      // 1. You can return { status: 'completed' } immediately for fast tasks
      // 2. Or return { status: 'in-progress' } for long-running tasks
      const taskId = input.params.id;

      // Simulate starting a background task
      tasks.set(taskId, { status: "running" });

      setTimeout(() => {
        tasks.set(taskId, { status: "completed", result: "Search results" });
      }, 5000);

      return {
        result: {
          status: "in-progress",
          details: "Starting web search...",
        },
      };
    },
    fetchToolResult: async (input) => {
      // Polled by the pipeline if executeTool returned 'in-progress'
      const task = tasks.get(input.params.id);

      if (task?.status === "completed") {
        tasks.delete(input.params.id);
        return {
          result: {
            status: "completed",
            result: [{ type: "text", text: task.result }],
          },
        };
      }

      return {
        result: {
          status: "in-progress",
          details: "Still searching...",
        },
      };
    },
    onToolCanceled: (input) => {
      // Cleanup if the model or user cancels the run
      tasks.delete(input.params.id);
    },
  };
});

// Bind to a transport and start serving
const transport = new StdioTransportServer();
await server.serve(transport);

Step 3: Register in the Pipeline

const pipeline = new LmPipeline({
  model: myModel,
  extensions: [
    new LmPipelineExtensionClient({
      transport: new StdioTransportClient("node search-extension.js"),
    }),
  ],
});