vendor_connectors.registry
Vendor Connector Registry with Entry Points.
This module provides automatic discovery and registration of vendor connectors using Python’s entry points system. This allows:
- DRY interface via VendorConnectorBase ABC
- Automatic discovery of all connectors (even from other packages)
- Unified factory function for instantiation
- Same registry used by both MCP and CLI
Usage: from vendor_connectors.registry import get_connector, list_connectors
# List available connectorsavailable = list_connectors()# {'jules': <class JulesConnector>, 'cursor': <class CursorConnector>, ...}
# Get a specific connector instanceconnector = get_connector('jules', api_key='...')
# Use itsources = connector.list_sources()Entry Points (in pyproject.toml): [project.entry-points.”vendor_connectors.connectors”] jules = “vendor_connectors.google.jules:JulesConnector” cursor = “vendor_connectors.cursor:CursorConnector” github = “vendor_connectors.github:GitHubConnector”
Module Contents
Section titled “Module Contents”Functions
Section titled “Functions”_discover_connectors | Discover all registered connectors via entry points. |
|---|---|
_register_builtins | Register built-in connectors that may not be in entry points yet. |
list_connectors | List all available connectors. |
get_connector_class | Get a connector class by name. |
get_connector | Factory to instantiate a connector by name. |
clear_cache | Clear the connector cache (useful for testing). |
get_connector_info | Get metadata about a connector. |
list_connector_info | Get metadata for all connectors. |
_connector_cache |
|---|
vendor_connectors.registry._connector_cache : dict[str, type[vendor_connectors.base.VendorConnectorBase]] | None
Section titled “vendor_connectors.registry._connector_cache : dict[str, type[vendor_connectors.base.VendorConnectorBase]] | None”None
vendor_connectors.registry._discover_connectors() → dict[str, type[vendor_connectors.base.VendorConnectorBase]]
Section titled “vendor_connectors.registry._discover_connectors() → dict[str, type[vendor_connectors.base.VendorConnectorBase]]”Discover all registered connectors via entry points.
vendor_connectors.registry._register_builtins(connectors: dict[str, type[vendor_connectors.base.VendorConnectorBase]]) → None
Section titled “vendor_connectors.registry._register_builtins(connectors: dict[str, type[vendor_connectors.base.VendorConnectorBase]]) → None”Register built-in connectors that may not be in entry points yet.
vendor_connectors.registry.list_connectors() → dict[str, type[vendor_connectors.base.VendorConnectorBase]]
Section titled “vendor_connectors.registry.list_connectors() → dict[str, type[vendor_connectors.base.VendorConnectorBase]]”List all available connectors.
Returns: Dict mapping connector name to connector class.
vendor_connectors.registry.get_connector_class(name: str) → type[vendor_connectors.base.VendorConnectorBase]
Section titled “vendor_connectors.registry.get_connector_class(name: str) → type[vendor_connectors.base.VendorConnectorBase]”Get a connector class by name.
Args: name: Connector name (e.g., ‘jules’, ‘cursor’, ‘github’)
Returns: The connector class.
Raises: ValueError: If connector not found.
vendor_connectors.registry.get_connector(name: str, **kwargs: Any) → vendor_connectors.base.VendorConnectorBase
Section titled “vendor_connectors.registry.get_connector(name: str, **kwargs: Any) → vendor_connectors.base.VendorConnectorBase”Factory to instantiate a connector by name.
Args: name: Connector name (e.g., ‘jules’, ‘cursor’, ‘github’) **kwargs: Arguments passed to connector constructor
Returns: Instantiated connector.
Raises: ValueError: If connector not found.
Example:
connector = get_connector(‘jules’, api_key=’…’) connector.list_sources()
vendor_connectors.registry.clear_cache() → None
Section titled “vendor_connectors.registry.clear_cache() → None”Clear the connector cache (useful for testing).
vendor_connectors.registry.get_connector_info(name: str) → dict[str, Any]
Section titled “vendor_connectors.registry.get_connector_info(name: str) → dict[str, Any]”Get metadata about a connector.
Returns: Dict with name, module, env_vars, description, etc.
vendor_connectors.registry.list_connector_info() → list[dict[str, Any]]
Section titled “vendor_connectors.registry.list_connector_info() → list[dict[str, Any]]”Get metadata for all connectors.