Skip to content

Tutorial: Create a Custom Component

This tutorial walks through building a new G6 component from scratch, including Input/Output models, an AIBlock subclass, optional MCP tools, and tests.

What you will build

A text_summarizer component that summarizes text documents with configurable strategies. The component will follow G6's standard patterns:

  • Pydantic Input/Output models
  • AIBlock subclass with infer method
  • Result[T] railway error handling
  • Optional MCP sub-package

Step 1: Component structure

Create the component directory:

components/mvp/text_summarizer/
├── __init__.py          # Public exports
├── schema.py            # Pydantic Input/Output models
└── block.py             # AIBlock subclass

Loose theme

G6 uses the Polylith loose theme. Components live at components/mvp/<name>/, not components/<name>/mvp/<name>/.

Step 2: Define Input/Output models

Create schema.py with Pydantic models:

# components/mvp/text_summarizer/schema.py
"""Input/Output schemas for the text_summarizer component."""

from pydantic import BaseModel, Field


class SummaryInput(BaseModel):
    """Input for text summarization."""

    text: str = Field(..., description="Text to summarize")
    strategy: str = Field(
        default="extractive",
        description="Summarization strategy: 'extractive' or 'truncate'",
    )
    max_sentences: int = Field(
        default=3,
        description="Maximum sentences in summary",
    )
    operation: str = Field(
        default="summarize",
        description="Operation: 'summarize' or 'keywords'",
    )


class SummaryOutput(BaseModel):
    """Output from text summarization."""

    summary: str = Field(..., description="Summarized text")
    sentence_count: int = Field(..., description="Number of sentences in summary")
    compression_ratio: float = Field(
        ..., description="Ratio of summary length to original"
    )
    keywords: list[str] = Field(
        default_factory=list, description="Extracted keywords"
    )

Design conventions

  • Use Field(...) for required fields and Field(default=...) for optional ones
  • Include description on every field for MCP tool documentation
  • Keep models frozen when possible (use model_config = ConfigDict(frozen=True) for immutable inputs)
  • Include an operation field if the component supports multiple operations

Step 3: Implement the AIBlock subclass

Create block.py:

# components/mvp/text_summarizer/block.py
"""Text summarization block."""

from __future__ import annotations

import re
from collections import Counter
from dataclasses import dataclass, field

from mvp.core import AIBlock, Result
from mvp.text_summarizer.schema import SummaryInput, SummaryOutput


@dataclass
class TextSummarizerBlock(AIBlock[SummaryInput, SummaryOutput, dict]):
    """Summarizes text documents using extractive or truncation strategies."""

    name: str = "text_summarizer"
    state: dict = field(default_factory=dict)

    def infer(self, input: SummaryInput) -> Result[SummaryOutput]:
        """Run summarization based on the requested operation."""
        ops = {
            "summarize": self._summarize,
            "keywords": self._extract_keywords,
        }

        handler = ops.get(input.operation)
        if handler is None:
            return Result.fail(
                f"[TEXT_SUMMARIZER] Unknown operation: {input.operation}"
            )

        try:
            return handler(input)
        except Exception as e:
            return Result.fail(f"[TEXT_SUMMARIZER] {e}")

    def _summarize(self, input: SummaryInput) -> Result[SummaryOutput]:
        """Summarize text using the specified strategy."""
        if not input.text.strip():
            return Result.fail("[TEXT_SUMMARIZER] Empty input text")

        if input.strategy == "extractive":
            summary = self._extractive_summary(input.text, input.max_sentences)
        elif input.strategy == "truncate":
            summary = self._truncate_summary(input.text, input.max_sentences)
        else:
            return Result.fail(
                f"[TEXT_SUMMARIZER] Unknown strategy: {input.strategy}"
            )

        sentences = self._split_sentences(summary)
        ratio = len(summary) / len(input.text) if input.text else 0.0

        return Result.ok(SummaryOutput(
            summary=summary,
            sentence_count=len(sentences),
            compression_ratio=round(ratio, 3),
        ))

    def _extract_keywords(self, input: SummaryInput) -> Result[SummaryOutput]:
        """Extract keywords from text."""
        words = re.findall(r'\b[a-zA-Z]{4,}\b', input.text.lower())
        # Simple stopword filtering
        stopwords = {"this", "that", "with", "from", "have", "been", "will", "would"}
        filtered = [w for w in words if w not in stopwords]
        counts = Counter(filtered)
        keywords = [word for word, _ in counts.most_common(10)]

        return Result.ok(SummaryOutput(
            summary=input.text[:200],
            sentence_count=0,
            compression_ratio=0.0,
            keywords=keywords,
        ))

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

    def _extractive_summary(self, text: str, max_sentences: int) -> str:
        """Score sentences by word frequency and select top ones."""
        sentences = self._split_sentences(text)
        if len(sentences) <= max_sentences:
            return text

        # Score by word frequency
        words = re.findall(r'\b\w+\b', text.lower())
        freq = Counter(words)

        scored = []
        for i, sent in enumerate(sentences):
            sent_words = re.findall(r'\b\w+\b', sent.lower())
            score = sum(freq.get(w, 0) for w in sent_words)
            scored.append((i, score, sent))

        # Select top sentences, maintain original order
        top = sorted(scored, key=lambda x: x[1], reverse=True)[:max_sentences]
        top_ordered = sorted(top, key=lambda x: x[0])

        return ". ".join(s for _, _, s in top_ordered) + "."

    def _truncate_summary(self, text: str, max_sentences: int) -> str:
        """Simply take the first N sentences."""
        sentences = self._split_sentences(text)
        selected = sentences[:max_sentences]
        return ". ".join(selected) + "." if selected else text

Key patterns

  • Error prefix -- use [COMPONENT_NAME] in error messages for easy log filtering
  • Operation dispatch -- use a dict mapping operation names to handler methods
  • Result.fail for errors -- never raise exceptions; wrap them in Result.fail
  • Result.ok for success -- always return structured output via Result.ok
  • state: dict -- use field(default_factory=dict) for mutable state
  • __post_init__ -- initialize state in __post_init__ if needed

Step 4: Export from __init__.py

# components/mvp/text_summarizer/__init__.py
"""Text summarization component."""

from mvp.text_summarizer.schema import SummaryInput, SummaryOutput
from mvp.text_summarizer.block import TextSummarizerBlock

__all__ = ["SummaryInput", "SummaryOutput", "TextSummarizerBlock"]

The ComponentRegistry auto-discovers this component at import time.

Step 5: Add MCP sub-package (optional)

For MCP integration, create a sub-package:

components/mvp/text_summarizer/
├── __init__.py
├── schema.py
├── block.py
└── text_summarizer_mcp/
    ├── __init__.py
    └── server.py

The MCP server exposes component operations as tools:

# components/mvp/text_summarizer/text_summarizer_mcp/server.py
"""MCP tools for text_summarizer."""

from fastmcp import FastMCP

server = FastMCP("text_summarizer")


@server.tool()
def summarize_text(
    text: str,
    strategy: str = "extractive",
    max_sentences: int = 3,
) -> str:
    """Summarize a text document.

    Args:
        text: Text to summarize.
        strategy: 'extractive' or 'truncate'.
        max_sentences: Maximum sentences in summary.
    """
    from mvp.text_summarizer import TextSummarizerBlock, SummaryInput

    block = TextSummarizerBlock()
    result = block.infer(SummaryInput(
        text=text,
        strategy=strategy,
        max_sentences=max_sentences,
    ))

    if result.is_ok():
        return result.value.summary
    return f"Error: {result.error}"


@server.tool()
def extract_keywords(text: str) -> str:
    """Extract keywords from text.

    Args:
        text: Text to extract keywords from.
    """
    from mvp.text_summarizer import TextSummarizerBlock, SummaryInput

    block = TextSummarizerBlock()
    result = block.infer(SummaryInput(text=text, operation="keywords"))

    if result.is_ok():
        return ", ".join(result.value.keywords)
    return f"Error: {result.error}"

Lazy imports in MCP tools

Import component modules inside tool functions (not at module level) to avoid circular imports and reduce startup time.

Step 6: Register in workspace

The component is automatically discovered by the ComponentRegistry. Verify:

from mvp.core.registry import get_registry

registry = get_registry()
meta = [c for c in registry.list_all() if c.name == "text_summarizer"]
print(meta)  # [ComponentMeta(name='text_summarizer', ...)]

Step 7: Write tests

Create tests/mvp/text_summarizer/test_text_summarizer.py:

"""Tests for text_summarizer component."""

import pytest
from mvp.text_summarizer import SummaryInput, SummaryOutput, TextSummarizerBlock


@pytest.fixture
def block():
    return TextSummarizerBlock()


@pytest.fixture
def sample_text():
    return (
        "Python is a programming language. It is widely used for web development. "
        "Python supports multiple paradigms. It has a large standard library. "
        "Many data scientists use Python for analysis."
    )


class TestTextSummarizerBlock:
    def test_extractive_summary(self, block, sample_text):
        result = block.infer(SummaryInput(text=sample_text, max_sentences=2))
        assert result.is_ok()
        assert result.value.sentence_count <= 2
        assert result.value.compression_ratio < 1.0

    def test_truncate_summary(self, block, sample_text):
        result = block.infer(SummaryInput(
            text=sample_text, strategy="truncate", max_sentences=2
        ))
        assert result.is_ok()
        assert result.value.sentence_count <= 2

    def test_keywords(self, block, sample_text):
        result = block.infer(SummaryInput(
            text=sample_text, operation="keywords"
        ))
        assert result.is_ok()
        assert len(result.value.keywords) > 0
        assert "python" in result.value.keywords

    def test_empty_text(self, block):
        result = block.infer(SummaryInput(text=""))
        assert result.is_fail()
        assert "Empty input" in result.error

    def test_unknown_operation(self, block, sample_text):
        result = block.infer(SummaryInput(text=sample_text, operation="invalid"))
        assert result.is_fail()
        assert "Unknown operation" in result.error

    def test_unknown_strategy(self, block, sample_text):
        result = block.infer(SummaryInput(
            text=sample_text, strategy="invalid"
        ))
        assert result.is_fail()
        assert "Unknown strategy" in result.error

    def test_pipeline_composition(self, block):
        """Test that the block can compose via >>."""
        pipeline = block >> block  # Summarize the summary
        # PipelineBlock is created successfully
        assert pipeline is not None

Run tests:

python -m pytest tests/mvp/text_summarizer/ -v

Component checklist

  • schema.py with Pydantic Input/Output models
  • block.py with AIBlock subclass and infer method
  • __init__.py with __all__ exports
  • Error handling via Result.fail (no raised exceptions)
  • Error messages prefixed with [COMPONENT_NAME]
  • Operation dispatch via dict mapping
  • MCP sub-package (optional)
  • Tests covering success, failure, and edge cases

Next steps