src.core.service_registry module#

Marcus Service Registry.

Manages service advertisement and discovery for Marcus instances. Allows multiple clients (Cato, Claude Desktop, etc.) to discover and connect to running Marcus instances.

class src.core.service_registry.MarcusServiceRegistry[source]#

Bases: object

Manages Marcus service advertisement and discovery.

When Marcus starts, it registers itself in a discoverable location. Clients like Cato can find running Marcus instances automatically.

__init__(instance_id=None)[source]#

Initialize service registry.

Parameters:

instance_id (Optional[str]) – Unique identifier for this Marcus instance

register_service(mcp_command, log_dir, project_name=None, provider=None, **kwargs)[source]#

Register this Marcus instance as available service.

Parameters:

  • mcp_command (str) – Command to connect to this Marcus MCP server

  • log_dir (str) – Directory where Marcus writes logs

  • project_name (Optional[str]) – Current project name

  • provider (Optional[str]) – Kanban provider being used

  • **kwargs (Any) – Additional service metadata

Returns:

Service registration info

Return type:

Dict[str, Any]

update_heartbeat(**updates)[source]#

Update service heartbeat and optional fields.

Return type:

None

Parameters:

updates (Any)

unregister_service()[source]#

Remove service registration.

Return type:

None

classmethod discover_services()[source]#

Discover all running Marcus services.

Returns:

List of available Marcus services

Return type:

List[Dict[str, Any]]

classmethod get_preferred_service()[source]#

Get the preferred Marcus service to connect to.

Returns most recently started service, or None if none available.

Returns:

Preferred service info, or None

Return type:

Optional[Dict[str, Any]]

src.core.service_registry.get_service_registry(instance_id=None)[source]#

Get or create global service registry instance.

Return type:

MarcusServiceRegistry

Parameters:

instance_id (str | None)

src.core.service_registry.register_marcus_service(**kwargs)[source]#

Register Marcus service.

Return type:

Dict[str, Any]

Parameters:

kwargs (Any)

src.core.service_registry.unregister_marcus_service()[source]#

Unregister Marcus service.

Return type:

None