carboncue_sdk.client

client

Main client for CarbonCue SDK.

CarbonClient

CarbonClient(config: CarbonConfig | None = None)

Client for accessing carbon intensity data and calculating SCI scores.

Example

client = CarbonClient() intensity = await client.get_current_intensity(region="us-west-2") sci = client.calculate_sci( ... operations=100.0, ... materials=50.0, ... functional_unit=1000, ... region="us-west-2" ... )

Parameters:

Name	Type	Description	Default
config	CarbonConfig | None	Optional configuration. If not provided, loads from environment.	None

Source code in packages/sdk/src/carboncue_sdk/client.py

def __init__(self, config: CarbonConfig | None = None) -> None:
    """Initialize the carbon client.

    Args:
        config: Optional configuration. If not provided, loads from environment.
    """
    self.config = config or CarbonConfig()
    self._http_client: httpx.AsyncClient | None = None
    self._cache: dict[str, tuple[Any, datetime]] = {}

__aenter__ async

__aenter__() -> CarbonClient

Async context manager entry.

Source code in packages/sdk/src/carboncue_sdk/client.py

async def __aenter__(self) -> "CarbonClient":
    """Async context manager entry."""
    self._http_client = httpx.AsyncClient(
        timeout=self.config.request_timeout,
        headers={"auth-token": self.config.electricity_maps_api_key or ""},
    )
    return self

__aexit__ async

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Async context manager exit.

Source code in packages/sdk/src/carboncue_sdk/client.py

async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Async context manager exit."""
    if self._http_client:
        await self._http_client.aclose()
        self._http_client = None

get_current_intensity async

get_current_intensity(region: str, provider: str = 'aws') -> CarbonIntensity

Get current carbon intensity for a region.

Parameters:

Name	Type	Description	Default
region	str	Region code (e.g., us-west-2)	required
provider	str	Cloud provider (aws, azure, gcp, etc.)	'aws'

Returns:

Type	Description
CarbonIntensity	Current carbon intensity data

Raises:

Type	Description
InvalidRegionError	If region is not supported
InvalidProviderError	If provider is not supported
AuthenticationError	If API key is missing or invalid
RateLimitError	If API rate limit exceeded
DataNotAvailableError	If data not available for region
APIError	If API request fails

Source code in packages/sdk/src/carboncue_sdk/client.py

async def get_current_intensity(self, region: str, provider: str = "aws") -> CarbonIntensity:
    """Get current carbon intensity for a region.

    Args:
        region: Region code (e.g., us-west-2)
        provider: Cloud provider (aws, azure, gcp, etc.)

    Returns:
        Current carbon intensity data

    Raises:
        InvalidRegionError: If region is not supported
        InvalidProviderError: If provider is not supported
        AuthenticationError: If API key is missing or invalid
        RateLimitError: If API rate limit exceeded
        DataNotAvailableError: If data not available for region
        APIError: If API request fails
    """
    cache_key = f"intensity:{provider}:{region}"
    cached = self._get_from_cache(cache_key)
    if cached is not None:
        return cached  # type: ignore[no-any-return]

    # Map cloud region to Electricity Maps zone
    try:
        zone_id = RegionMapper.get_zone_id(region, provider)
    except ValueError as e:
        error_msg = str(e)
        if "cloud provider" in error_msg.lower():
            raise InvalidProviderError(error_msg) from e
        raise InvalidRegionError(error_msg) from e

    # Get data from Electricity Maps API
    intensity = await self._fetch_from_electricity_maps(zone_id, region)

    self._set_cache(cache_key, intensity)
    return intensity

calculate_sci

calculate_sci(operational_emissions: float, embodied_emissions: float, functional_unit: float, functional_unit_type: str = 'requests', region: str = 'us-west-2') -> SCIScore

Calculate Software Carbon Intensity (SCI) score.

Implements GSF SCI specification: SCI = (O + M) / R

Parameters:

Name	Type	Description	Default
operational_emissions	float	O - Operational emissions in gCO2eq	required
embodied_emissions	float	M - Embodied emissions in gCO2eq	required
functional_unit	float	R - Number of functional units	required
functional_unit_type	str	Type of functional unit (requests, users, etc.)	'requests'
region	str	Region where computation occurred	'us-west-2'

Returns:

Type	Description
SCIScore	Calculated SCI score

Raises:

Type	Description
ValueError	If functional_unit is <= 0

Source code in packages/sdk/src/carboncue_sdk/client.py

def calculate_sci(
    self,
    operational_emissions: float,
    embodied_emissions: float,
    functional_unit: float,
    functional_unit_type: str = "requests",
    region: str = "us-west-2",
) -> SCIScore:
    """Calculate Software Carbon Intensity (SCI) score.

    Implements GSF SCI specification: SCI = (O + M) / R

    Args:
        operational_emissions: O - Operational emissions in gCO2eq
        embodied_emissions: M - Embodied emissions in gCO2eq
        functional_unit: R - Number of functional units
        functional_unit_type: Type of functional unit (requests, users, etc.)
        region: Region where computation occurred

    Returns:
        Calculated SCI score

    Raises:
        ValueError: If functional_unit is <= 0
    """
    if functional_unit <= 0:
        raise ValueError("Functional unit must be greater than 0")

    score = (operational_emissions + embodied_emissions) / functional_unit

    return SCIScore(
        score=score,
        operational_emissions=operational_emissions,
        embodied_emissions=embodied_emissions,
        functional_unit=functional_unit,
        functional_unit_type=functional_unit_type,
        region=region,
    )

close async

close() -> None

Close the HTTP client and cleanup resources.

Source code in packages/sdk/src/carboncue_sdk/client.py

async def close(self) -> None:
    """Close the HTTP client and cleanup resources."""
    if self._http_client:
        await self._http_client.aclose()
        self._http_client = None