Skip to content

Tutorial: Create a Custom Component

G6 ships with 200+ built-in components, but you can add your own with the extension system. An extension is a small folder you drop into ~/.g6/extensions/ — G6 discovers it at runtime, exposes it across the surfaces (MCP, REST, CLI), runs it in a sandbox, and hot-reloads it on change. You build on the installed app; you don't modify G6's source.

Extensions vs the core source

Extensions live in the g6ext.* namespace, kept separate from G6's core mvp.* modules for security. Your extension imports the SDK primitives it needs from the installed runtime (mvp.core.ai_block, mvp.core.result) but ships and runs as your own sandboxed package.

What you will build

A text_summarizer extension that summarizes text with configurable strategies, exposed as a component and as MCP tools. The folder layout:

~/.g6/extensions/text_summarizer/
├── manifest.json     # Capabilities, tier, sandbox level
├── __init__.py       # Marks the package
├── block.py          # AIBlock implementation
├── tools.py          # MCP tool registration (optional)
└── routes.py         # REST routes (optional)

Step 1: Write the manifest

manifest.json tells G6 what your extension is, which modules to load, and how to sandbox it:

{
  "name": "text_summarizer",
  "version": "1.0.0",
  "description": "Summarize text with extractive or truncation strategies",
  "author": "Your Name",
  "block_module": "block",
  "tools_module": "tools",
  "operations": ["summarize", "keywords"],
  "mcp_tool_names": ["text_summarizer_summarize", "text_summarizer_keywords"],
  "required_components": ["core"],
  "tier": "premium",
  "sandbox": "restricted",
  "learning_enabled": false
}
Field Meaning
block_module / tools_module Python modules in the folder that hold the block and MCP tools
operations Operation names your block handles
mcp_tool_names Tools your extension advertises to MCP clients
required_components Built-in components your extension depends on
tier Minimum license tier to load the extension
sandbox restricted, standard, or trusted — isolation level for the subprocess

Step 2: Implement the block

block.py subclasses AIBlock and returns a Result — never raise; wrap errors in Result.fail:

# ~/.g6/extensions/text_summarizer/block.py
from __future__ import annotations

import re
from collections import Counter
from dataclasses import dataclass

from mvp.core.ai_block import AIBlock
from mvp.core.result import Result


@dataclass
class TextSummarizerBlock(AIBlock):
    """Summarizes text with extractive or truncation strategies."""

    name: str = "text_summarizer"

    def infer(self, data):
        op = data.get("operation", "summarize") if isinstance(data, dict) else "summarize"
        if op == "summarize":
            return self._summarize(data)
        if op == "keywords":
            return self._keywords(data)
        return Result.fail(f"[TEXT_SUMMARIZER] Unknown operation: {op}")

    def _summarize(self, data: dict) -> Result:
        text = (data.get("text") or "").strip()
        if not text:
            return Result.fail("[TEXT_SUMMARIZER] Empty input text")
        max_sentences = int(data.get("max_sentences", 3))
        strategy = data.get("strategy", "extractive")
        if strategy == "extractive":
            summary = self._extractive(text, max_sentences)
        elif strategy == "truncate":
            summary = ". ".join(self._sentences(text)[:max_sentences]) + "."
        else:
            return Result.fail(f"[TEXT_SUMMARIZER] Unknown strategy: {strategy}")
        ratio = round(len(summary) / len(text), 3)
        return Result.ok({"summary": summary, "compression_ratio": ratio})

    def _keywords(self, data: dict) -> Result:
        words = re.findall(r"\b[a-zA-Z]{4,}\b", (data.get("text") or "").lower())
        stop = {"this", "that", "with", "from", "have", "been", "will", "would"}
        counts = Counter(w for w in words if w not in stop)
        return Result.ok({"keywords": [w for w, _ in counts.most_common(10)]})

    def _sentences(self, text: str) -> list[str]:
        return [s.strip() for s in re.split(r"[.!?]+", text) if s.strip()]

    def _extractive(self, text: str, max_sentences: int) -> str:
        sentences = self._sentences(text)
        if len(sentences) <= max_sentences:
            return text
        freq = Counter(re.findall(r"\b\w+\b", text.lower()))
        scored = [
            (i, sum(freq.get(w, 0) for w in re.findall(r"\b\w+\b", s.lower())), s)
            for i, s in enumerate(sentences)
        ]
        top = sorted(sorted(scored, key=lambda x: x[1], reverse=True)[:max_sentences])
        return ". ".join(s for _, _, s in top) + "."

Mark the package with an __init__.py:

# ~/.g6/extensions/text_summarizer/__init__.py
"""text_summarizer extension"""

Key patterns

  • Error prefix — start error messages with [COMPONENT_NAME] for easy log filtering
  • Operation dispatch — branch on an operation key to support multiple operations
  • Result.fail for errors — never raise; wrap exceptions in Result.fail
  • Result.ok for success — always return structured output via Result.ok

Step 3: Expose MCP tools (optional)

tools.py defines a register(server) function that adds your tools to the MCP server. Import your block lazily inside each tool to keep startup fast:

# ~/.g6/extensions/text_summarizer/tools.py
import json


def register(server):
    @server.tool()
    def text_summarizer_summarize(text: str, strategy: str = "extractive", max_sentences: int = 3) -> str:
        """Summarize a text document."""
        from g6ext.text_summarizer.block import TextSummarizerBlock

        result = TextSummarizerBlock().infer(
            {"operation": "summarize", "text": text, "strategy": strategy, "max_sentences": max_sentences}
        )
        return json.dumps(result.value if result.is_ok() else {"error": result.error})

    @server.tool()
    def text_summarizer_keywords(text: str) -> str:
        """Extract keywords from text."""
        from g6ext.text_summarizer.block import TextSummarizerBlock

        result = TextSummarizerBlock().infer({"operation": "keywords", "text": text})
        return json.dumps(result.value if result.is_ok() else {"error": result.error})

To also expose REST routes, add a routes.py and list it in your manifest — see the Extension SDK reference for the routing API.

Step 4: Install and verify

Drop the folder into ~/.g6/extensions/ (or your configured extensions directory). G6 loads it on the next start and hot-reloads on file changes — no rebuild needed. Verify it is live:

List all available G6 components

text_summarizer appears in the list, and your text_summarizer_summarize / text_summarizer_keywords tools are now callable.

Open Settings → Diagnostics to confirm the extension loaded, then use it from the Ask G6 chat or any run.

Sandbox levels

Extensions run in a subprocess sandbox by default. Use restricted while developing, and only raise to standard or trusted when your extension genuinely needs broader access.

Step 5: Test your block

Test the block directly with pytest:

# ~/.g6/extensions/text_summarizer/test_block.py
from g6ext.text_summarizer.block import TextSummarizerBlock


def test_summarize():
    result = TextSummarizerBlock().infer(
        {"operation": "summarize", "text": "A. B. C. D. E.", "max_sentences": 2}
    )
    assert result.is_ok()
    assert result.value["compression_ratio"] < 1.0


def test_unknown_operation():
    result = TextSummarizerBlock().infer({"operation": "nope", "text": "x"})
    assert result.is_fail()
    assert "Unknown operation" in result.error

Extension checklist

  • manifest.json with name, block_module, operations, tier, and sandbox
  • block.py with an AIBlock subclass and an infer method returning Result
  • __init__.py marking the package
  • Error handling via Result.fail (no raised exceptions), messages prefixed [COMPONENT_NAME]
  • tools.py with register(server) (optional MCP), routes.py (optional REST)
  • Tests covering success, failure, and edge cases

Next steps