cycloid-mcp-components

Cycloid MCP Component Development Rules

Component Architecture Overview

Components are plain async functions in src/components/, auto-discovered by FileSystemProvider. No classes, no inheritance — just decorated functions with dependency injection.

Component Structure

Each component is a single file in src/components/:

src/components/
├── catalogs.py      # @tool + @resource functions
├── events.py        # @tool + @resource functions
├── pipelines.py     # @tool + @resource functions
├── stacks.py        # @tool + @resource + elicitation
└── stackforms.py    # @tool function

Component Implementation Pattern

Tool + Resource Template

"""Feature tools and resources for Cycloid MCP server."""

import json
from typing import Any, Dict

from fastmcp.dependencies import Depends  # type: ignore[reportAttributeAccessIssue]
from fastmcp.exceptions import ToolError
from fastmcp.resources import resource
from fastmcp.tools import tool
from fastmcp.utilities.logging import get_logger

from src.cli import CLIMixin
from src.dependencies import get_cli
from src.exceptions import CycloidCLIError
from src.types import JSONList

logger = get_logger(__name__)


async def _get_items(cli: CLIMixin) -> JSONList:
    """Private helper — get items from CLI."""
    data = await cli.execute_cli("command", ["list"])
    return cli.process_cli_response(data, list_key="items")


@tool(
    name="CYCLOID_ITEM_LIST",
    description="List all items with their details.",
    annotations={"readOnlyHint": True},
)
async def list_items(
    cli: CLIMixin = Depends(get_cli),  # type: ignore[reportCallInDefaultInitializer]
) -> Dict[str, Any]:
    """List all items. Returns dict with items list and count."""
    try:
        items = await _get_items(cli)
        return {"items": items, "count": len(items)}
    except CycloidCLIError as e:
        raise ToolError(f"Failed to list items: {str(e)}")
    except Exception as e:
        raise ToolError(f"Error listing items: {str(e)}")


@resource("cycloid://items")
async def get_items_resource(
    cli: CLIMixin = Depends(get_cli),  # type: ignore[reportCallInDefaultInitializer]
) -> str:
    """Get all items as a resource."""
    try:
        items = await _get_items(cli)
        return json.dumps({"items": items, "count": len(items)}, indent=2)
    except CycloidCLIError as e:
        logger.error(f"Failed to read items resource: {str(e)}")
        return json.dumps({"error": str(e), "items": [], "count": 0}, indent=2)