Coverage for integrations / agent_engine / revenue_tools.py: 0.0%
53 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Revenue Agent Tools — AutoGen tools for API revenue monitoring and optimization.
4Tier 2 tools (agent_engine context). Same registration pattern as marketing_tools.py.
5"""
6import json
7import logging
8from typing import Annotated, Optional
10logger = logging.getLogger('hevolve_social')
13def register_revenue_tools(helper, assistant, user_id: str):
14 """Register revenue tools with an AutoGen agent (Tier 2)."""
16 def get_api_revenue_stats() -> str:
17 """Get total API revenue, top endpoints, growth trends, tier distribution."""
18 try:
19 from integrations.social.models import get_db
20 from integrations.agent_engine.commercial_api import CommercialAPIService
21 from integrations.social.models import CommercialAPIKey, APIUsageLog
22 from sqlalchemy import func
24 db = get_db()
25 try:
26 total_keys = db.query(CommercialAPIKey).filter_by(is_active=True).count()
27 tier_dist = {}
28 for tier in ('free', 'starter', 'pro', 'enterprise'):
29 tier_dist[tier] = db.query(CommercialAPIKey).filter_by(
30 tier=tier, is_active=True).count()
32 total_revenue = db.query(
33 func.coalesce(func.sum(APIUsageLog.cost_credits), 0.0)
34 ).scalar()
36 total_calls = db.query(APIUsageLog).count()
38 # Top endpoints
39 top_endpoints = db.query(
40 APIUsageLog.endpoint,
41 func.count(APIUsageLog.id).label('call_count'),
42 func.sum(APIUsageLog.cost_credits).label('revenue'),
43 ).group_by(APIUsageLog.endpoint).order_by(
44 func.count(APIUsageLog.id).desc()
45 ).limit(5).all()
47 return json.dumps({
48 'total_api_keys': total_keys,
49 'tier_distribution': tier_dist,
50 'total_revenue_credits': round(float(total_revenue), 4),
51 'total_api_calls': total_calls,
52 'top_endpoints': [
53 {'endpoint': e, 'calls': c, 'revenue': round(float(r or 0), 4)}
54 for e, c, r in top_endpoints
55 ],
56 })
57 finally:
58 db.close()
59 except Exception as e:
60 return json.dumps({'error': str(e)})
62 def adjust_pricing(
63 tier: Annotated[str, "Tier to adjust: free|starter|pro|enterprise"],
64 new_cost_per_1k: Annotated[float, "New cost per 1k tokens in credits"],
65 reason: Annotated[str, "Justification for the change"],
66 ) -> str:
67 """Recommend a pricing adjustment. Does NOT auto-apply — logs recommendation."""
68 if tier == 'free' and new_cost_per_1k > 0:
69 return json.dumps({
70 'error': 'Free tier must always remain at 0 cost. '
71 'We do not gatekeep intelligence.'
72 })
74 return json.dumps({
75 'recommendation': {
76 'tier': tier,
77 'proposed_cost_per_1k_tokens': new_cost_per_1k,
78 'reason': reason,
79 'status': 'pending_review',
80 'note': 'Pricing changes require manual approval. '
81 'This recommendation has been logged.',
82 }
83 })
85 def generate_api_docs(
86 format: Annotated[str, "Output format: markdown or json"] = 'markdown',
87 ) -> str:
88 """Generate API documentation for the intelligence endpoints."""
89 endpoints = [
90 {'method': 'POST', 'path': '/api/v1/intelligence/chat',
91 'auth': 'X-API-Key', 'description': 'Chat with Hevolve AI intelligence',
92 'body': {'prompt': 'string (required)'}},
93 {'method': 'POST', 'path': '/api/v1/intelligence/analyze',
94 'auth': 'X-API-Key', 'description': 'Analyze documents with AI',
95 'body': {'document': 'string (required)', 'question': 'string'}},
96 {'method': 'POST', 'path': '/api/v1/intelligence/generate',
97 'auth': 'X-API-Key', 'description': 'Generate media (image/audio/video)',
98 'body': {'modality': 'string', 'prompt': 'string (required)'}},
99 {'method': 'GET', 'path': '/api/v1/intelligence/hivemind',
100 'auth': 'X-API-Key', 'description': 'Query collective hive knowledge',
101 'params': {'query': 'string (required)'}},
102 {'method': 'GET', 'path': '/api/v1/intelligence/usage',
103 'auth': 'X-API-Key', 'description': 'Get your API usage statistics',
104 'params': {'days': 'int (default 30)'}},
105 ]
107 if format == 'json':
108 return json.dumps({'endpoints': endpoints})
110 md = "# Hevolve AI Intelligence API\n\n"
111 md += "## Authentication\n"
112 md += "Pass your API key via the `X-API-Key` header.\n\n"
113 md += "## Endpoints\n\n"
114 for ep in endpoints:
115 md += f"### `{ep['method']} {ep['path']}`\n"
116 md += f"{ep['description']}\n\n"
117 return md
119 def promote_api(
120 campaign_name: Annotated[str, "Name for the promotional campaign"],
121 target_audience: Annotated[str, "Target audience description"],
122 channels: Annotated[str, "Comma-separated channels: platform,twitter,linkedin"],
123 ) -> str:
124 """Create a marketing campaign to promote the intelligence API."""
125 try:
126 from integrations.social.models import get_db
127 from integrations.social.campaign_service import CampaignService
129 db = get_db()
130 try:
131 result = CampaignService.create_campaign(
132 db, created_by=user_id,
133 name=campaign_name,
134 campaign_type='awareness',
135 description=f'API promotion targeting: {target_audience}',
136 target_communities=channels.split(','),
137 )
138 db.commit()
139 return json.dumps({'campaign_created': True, 'campaign': result})
140 finally:
141 db.close()
142 except Exception as e:
143 return json.dumps({
144 'campaign_created': False,
145 'note': f'Campaign service unavailable: {e}. '
146 f'Logged promotion intent for: {campaign_name}'
147 })
149 tools = [
150 ('get_api_revenue_stats',
151 'Get total API revenue, top endpoints, tier distribution, and growth trends',
152 get_api_revenue_stats),
153 ('adjust_pricing',
154 'Recommend a pricing adjustment for an API tier (does not auto-apply)',
155 adjust_pricing),
156 ('generate_api_docs',
157 'Generate API documentation in markdown or JSON format',
158 generate_api_docs),
159 ('promote_api',
160 'Create a marketing campaign to promote the intelligence API',
161 promote_api),
162 ]
164 for name, desc, func in tools:
165 helper.register_for_llm(name=name, description=desc)(func)
166 assistant.register_for_execution(name=name)(func)
168 logger.info(f"Registered {len(tools)} revenue tools for user {user_id}")