Skip to content

registry

registry

Introspective auto-registration of crew-dcs entry points as MCP tools.

Discovers async classmethod factories that take auth (the get_by_id / search-style entry points) on an entity class and registers each via the generic factory, tagged by its owning class + tier + classification. Bound instance methods (which presuppose a fetched entity) are excluded — they don't fit the (auth, **params) mold and are surfaced via workflows instead.

The manifest is metadata-only: per-callable {exclude, name, tier, classification}.

discover_classmethod_tools

discover_classmethod_tools(cls: type)

Yield (name, bound_classmethod) for async classmethods of cls taking auth.

Source code in src/crew_dcs/mcp_server/registry.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def discover_classmethod_tools(cls: type):
    """Yield (name, bound_classmethod) for async classmethods of `cls` taking `auth`."""
    for name, member in inspect.getmembers(cls):
        if name.startswith("_"):
            continue
        # inspect.ismethod is True only for classmethods (bound to the class);
        # instance methods accessed via the class are plain functions.
        if not inspect.ismethod(member):
            continue
        if not inspect.iscoroutinefunction(member):
            continue
        if "auth" not in inspect.signature(member).parameters:
            continue
        yield name, member

register_entity_class

register_entity_class(
    mcp: Any,
    cls: type,
    auth_resolver: AuthResolver,
    *,
    tag: str,
    manifest: dict[str, dict] | None = None,
    default_tier: str = "secondary",
    denylist: frozenset[str] = DEFAULT_DENYLIST
) -> list[Any]

Auto-register an entity class's classmethod factories. Returns the Tools.

Source code in src/crew_dcs/mcp_server/registry.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def register_entity_class(
    mcp: Any,
    cls: type,
    auth_resolver: AuthResolver,
    *,
    tag: str,
    manifest: dict[str, dict] | None = None,
    default_tier: str = "secondary",
    denylist: frozenset[str] = DEFAULT_DENYLIST,
) -> list[Any]:
    """Auto-register an entity class's classmethod factories. Returns the Tools."""
    manifest = manifest or {}
    registered: list[Any] = []
    for name, member in discover_classmethod_tools(cls):
        spec = manifest.get(name, {})
        if name in denylist or spec.get("exclude"):
            continue
        tool_name = spec.get("name", f"{tag}_{name}")
        tags = {tag, spec.get("tier", default_tier)}
        annotations = _annotations_for(spec.get("classification", "read"))
        tool = make_tool(
            member,
            auth_resolver,
            name=tool_name,
            tags=tags,
            annotations=annotations,
        )
        mcp.add_tool(tool)
        registered.append(tool)
    return registered