src.cost_tracking package#

Cost Tracking Module for Marcus.

Provides token-based cost tracking for AI usage across projects.

class src.cost_tracking.TokenTracker[source]#

Bases: object

Tracks AI token usage per project with real-time cost monitoring.

Features: - Per-project token tracking - Real-time spend rate calculation - Cost projections based on current usage - Token usage history and analytics

__init__(cost_per_1k_tokens=0.03)[source]#

Initialize token tracker.

Parameters:

cost_per_1k_tokens (float) – Cost per 1000 tokens (default $0.03 for Claude)

load_historical_data()[source]#

Load historical token usage data.

Return type:

None

save_data()[source]#

Persist token usage data.

Return type:

None

async track_tokens(project_id, input_tokens, output_tokens, model='claude-3-sonnet', metadata=None)[source]#

Track token usage for a project.

Parameters:

  • project_id (str) – Project identifier

  • input_tokens (int) – Number of input tokens

  • output_tokens (int) – Number of output tokens

  • model (str) – AI model used

  • metadata (Optional[Dict[str, Any]]) – Additional context (task_id, agent_id, etc.)

Returns:

Dict with usage stats and cost info

Return type:

Dict[str, Any]

get_project_stats(project_id)[source]#

Get comprehensive stats for a project.

Parameters:

project_id (str) – Project identifier

Returns:

Dict containing: - total_tokens: Total tokens used - total_cost: Total cost incurred - current_spend_rate: Tokens/hour over last 5 minutes - average_spend_rate: Overall tokens/hour - projected_cost: Estimated total cost at current rate - session_duration: Time since first token

Return type:

Dict[str, Any]

get_all_projects_summary()[source]#

Get summary of all tracked projects.

Return type:

Dict[str, Any]

async start_monitoring()[source]#

Start background monitoring task.

Return type:

None

async stop_monitoring()[source]#

Stop background monitoring.

Return type:

None

class src.cost_tracking.AIUsageMiddleware[source]#

Bases: object

Middleware that wraps AI provider calls to track token usage.

This intercepts all calls to AI providers (OpenAI, Anthropic, etc.) and tracks token consumption per project.

__init__()[source]#
Return type:

None

set_project_context(agent_id, project_id, task_id=None)[source]#

Set the current project context for an agent.

This should be called when an agent starts working on a project/task.

Return type:

None

Parameters:

  • agent_id (str)

  • project_id (str)

  • task_id (str | None)

clear_project_context(agent_id)[source]#

Clear project context when agent finishes.

Return type:

None

Parameters:

agent_id (str)

get_current_project(agent_id)[source]#

Get current project for an agent.

Return type:

Optional[str]

Parameters:

agent_id (str)

track_ai_usage(func)[source]#

Decorate AI provider methods to track token usage.

This wraps AI provider methods to capture token usage.

Return type:

Callable[..., Any]

Parameters:

func (Callable[[...], Any])

wrap_ai_provider(provider_instance)[source]#

Wrap an AI provider instance to track all its method calls.

This modifies the provider instance to track token usage on all methods that make API calls.

Return type:

Any

Parameters:

provider_instance (Any)

src.cost_tracking.track_project_tokens(project_id, agent_id=None)[source]#

Context manager to track AI tokens for a specific project.

Return type:

Any

Parameters:

  • project_id (str)

  • agent_id (str | None)

Usage:
with track_project_tokens(“project_123”, “agent_1”):

# All AI calls in this block will be tracked to project_123 await ai_engine.analyze(…)

Submodules#