|
| 1 | +""" |
| 2 | +distribution.py — HYDRA Distribution Agent Swarm (ROI-prioritized) |
| 3 | +================================================================== |
| 4 | +Distribution is the binding constraint on HYDRA's treasury: the endpoints |
| 5 | +already work, but revenue only arrives when *paying agents discover HYDRA*. |
| 6 | +This module organizes distribution into a small set of focused, role-based |
| 7 | +agents — coordinated by an orchestrator — and runs them in expected-ROI order. |
| 8 | +
|
| 9 | +Design philosophy (explicit, per the mandate): |
| 10 | + - HIGH ROI over volume. Each agent targets channels where agents that can |
| 11 | + actually *pay via x402* are most likely to discover and call HYDRA. |
| 12 | + - Thoughtful priority. Agents run in ROI order; the orchestrator does not |
| 13 | + spray every list — it works the channels that convert to paid calls first. |
| 14 | + - Honest degradation. Network/credentials may be absent; each agent reports |
| 15 | + whether it executed, was a no-op, or needs a credential — never pretends. |
| 16 | +
|
| 17 | +These agents wrap (and prioritize) the existing, battle-tested registration |
| 18 | +functions in agent_discovery.py and autonomous_marketing.py rather than |
| 19 | +duplicating them — one coordinated brain instead of scattered calls. |
| 20 | +
|
| 21 | +ROI ranking rationale (where do paying x402 agents actually look first?): |
| 22 | + 1. x402 ecosystem registries/marketplaces — agents here are payment-native. |
| 23 | + 2. MCP registries — MCP clients can be wired to pay; large, growing surface. |
| 24 | + 3. Discoverability (search + LLM manifests) — how autonomous agents find APIs. |
| 25 | + 4. Public API directories — broad reach, slower conversion. |
| 26 | + 5. Developer content (Dev.to / GitHub) — top-of-funnel, longest lag. |
| 27 | +""" |
| 28 | + |
| 29 | +from __future__ import annotations |
| 30 | + |
| 31 | +import asyncio |
| 32 | +import logging |
| 33 | +from datetime import datetime, timezone |
| 34 | +from typing import Any, Callable, Optional |
| 35 | + |
| 36 | +logger = logging.getLogger("hydra.distribution") |
| 37 | + |
| 38 | + |
| 39 | +class DistributionAgent: |
| 40 | + """ |
| 41 | + One focused distribution role. Subclasses (or instances built with an action |
| 42 | + callable) own a single channel category and report a structured result. |
| 43 | + """ |
| 44 | + |
| 45 | + def __init__( |
| 46 | + self, |
| 47 | + name: str, |
| 48 | + role: str, |
| 49 | + priority: int, |
| 50 | + roi_rationale: str, |
| 51 | + action: Callable[[], Any], |
| 52 | + is_async: bool = False, |
| 53 | + requires: Optional[list[str]] = None, |
| 54 | + ) -> None: |
| 55 | + self.name = name |
| 56 | + self.role = role |
| 57 | + self.priority = priority # lower = higher ROI / runs first |
| 58 | + self.roi_rationale = roi_rationale |
| 59 | + self._action = action |
| 60 | + self._is_async = is_async |
| 61 | + self.requires = requires or [] |
| 62 | + self.last_result: dict = {} |
| 63 | + |
| 64 | + async def run(self) -> dict: |
| 65 | + """Execute the agent's action, capturing success/no-op/credential gaps.""" |
| 66 | + started = datetime.now(timezone.utc).isoformat() |
| 67 | + try: |
| 68 | + if self._is_async: |
| 69 | + detail = await self._action() |
| 70 | + else: |
| 71 | + detail = await asyncio.get_event_loop().run_in_executor(None, self._action) |
| 72 | + self.last_result = { |
| 73 | + "agent": self.name, |
| 74 | + "role": self.role, |
| 75 | + "priority": self.priority, |
| 76 | + "executed": True, |
| 77 | + "ran_at": started, |
| 78 | + "detail": detail, |
| 79 | + } |
| 80 | + except Exception as exc: # noqa: BLE001 — distribution is best-effort, never fatal |
| 81 | + logger.warning("[DIST] %s failed (non-fatal): %s", self.name, exc) |
| 82 | + self.last_result = { |
| 83 | + "agent": self.name, |
| 84 | + "role": self.role, |
| 85 | + "priority": self.priority, |
| 86 | + "executed": False, |
| 87 | + "ran_at": started, |
| 88 | + "error": str(exc)[:200], |
| 89 | + "requires": self.requires, |
| 90 | + } |
| 91 | + return self.last_result |
| 92 | + |
| 93 | + |
| 94 | +class DistributionOrchestrator: |
| 95 | + """ |
| 96 | + Coordinates the distribution agents in ROI-priority order, aggregates a |
| 97 | + report, and exposes status. Run autonomously by the automaton heartbeat. |
| 98 | + """ |
| 99 | + |
| 100 | + def __init__(self, marketing: Optional[Any] = None) -> None: |
| 101 | + # Lazy imports so optional deps / import cycles never break startup. |
| 102 | + from .agent_discovery import ( |
| 103 | + register_with_discovery_services, |
| 104 | + ping_search_engines, |
| 105 | + verify_deployment_health, |
| 106 | + ) |
| 107 | + if marketing is None: |
| 108 | + from .autonomous_marketing import AutonomousMarketing |
| 109 | + marketing = AutonomousMarketing() |
| 110 | + self._marketing = marketing |
| 111 | + |
| 112 | + self.agents: list[DistributionAgent] = [ |
| 113 | + DistributionAgent( |
| 114 | + name="x402_ecosystem", |
| 115 | + role="Register on x402 registries & agent marketplaces", |
| 116 | + priority=1, |
| 117 | + roi_rationale="Agents here are payment-native — highest conversion to paid calls.", |
| 118 | + action=register_with_discovery_services, |
| 119 | + is_async=True, |
| 120 | + ), |
| 121 | + DistributionAgent( |
| 122 | + name="discoverability", |
| 123 | + role="Search + LLM discoverability (sitemap pings, manifests)", |
| 124 | + priority=3, |
| 125 | + roi_rationale="How autonomous/LLM agents locate APIs; compounding, low cost.", |
| 126 | + action=ping_search_engines, |
| 127 | + is_async=True, |
| 128 | + ), |
| 129 | + DistributionAgent( |
| 130 | + name="self_verification", |
| 131 | + role="Verify discovery manifests & deployment health each cycle", |
| 132 | + priority=2, |
| 133 | + roi_rationale="A broken manifest silently kills discovery — protect the funnel first.", |
| 134 | + action=verify_deployment_health, |
| 135 | + is_async=True, |
| 136 | + ), |
| 137 | + DistributionAgent( |
| 138 | + name="api_directories", |
| 139 | + role="Submit to public API directories (APIs.guru, public-apis, etc.)", |
| 140 | + priority=4, |
| 141 | + roi_rationale="Broad reach, slower conversion; idempotent best-effort.", |
| 142 | + action=self._marketing.submit_to_api_directories, |
| 143 | + requires=["GITHUB_PAT (for PR-based submissions)"], |
| 144 | + ), |
| 145 | + DistributionAgent( |
| 146 | + name="openapi_directories", |
| 147 | + role="Submit OpenAPI spec to API spec directories", |
| 148 | + priority=4, |
| 149 | + roi_rationale="Machine-readable spec surfaces HYDRA to API-indexing agents.", |
| 150 | + action=self._marketing.submit_openapi_to_directories, |
| 151 | + ), |
| 152 | + DistributionAgent( |
| 153 | + name="developer_content", |
| 154 | + role="Dev.to article + GitHub discussions (top-of-funnel)", |
| 155 | + priority=5, |
| 156 | + roi_rationale="Longest lag to revenue; builds durable inbound over time.", |
| 157 | + action=self._run_content, |
| 158 | + requires=["DEVTO_API_KEY", "GITHUB_PAT"], |
| 159 | + ), |
| 160 | + ] |
| 161 | + self.agents.sort(key=lambda a: a.priority) |
| 162 | + self._last_cycle: dict = {} |
| 163 | + |
| 164 | + def _run_content(self) -> dict: |
| 165 | + """Bundle the slower content channels into one agent action (sync).""" |
| 166 | + out: dict = {} |
| 167 | + for fn_name in ("publish_dev_to_article", "post_github_discussions", "autonomous_seo_content"): |
| 168 | + fn = getattr(self._marketing, fn_name, None) |
| 169 | + if callable(fn): |
| 170 | + try: |
| 171 | + out[fn_name] = fn() |
| 172 | + except Exception as exc: # noqa: BLE001 |
| 173 | + out[fn_name] = {"status": "error", "error": str(exc)[:120]} |
| 174 | + return out |
| 175 | + |
| 176 | + async def run_cycle(self) -> dict: |
| 177 | + """Run all agents in ROI-priority order; return an aggregated report.""" |
| 178 | + logger.info("[DIST] Running distribution cycle — %d agents (ROI-ordered).", len(self.agents)) |
| 179 | + results = [] |
| 180 | + for agent in self.agents: |
| 181 | + results.append(await agent.run()) |
| 182 | + executed = sum(1 for r in results if r.get("executed")) |
| 183 | + self._last_cycle = { |
| 184 | + "ran_at": datetime.now(timezone.utc).isoformat(), |
| 185 | + "agents_total": len(self.agents), |
| 186 | + "agents_executed": executed, |
| 187 | + "results": results, |
| 188 | + } |
| 189 | + logger.info("[DIST] Cycle complete — %d/%d agents executed.", executed, len(self.agents)) |
| 190 | + return self._last_cycle |
| 191 | + |
| 192 | + def get_status(self) -> dict: |
| 193 | + """ROI-ranked roster + last cycle summary, for /status surfacing.""" |
| 194 | + return { |
| 195 | + "strategy": "high-ROI distribution (paying-agent reach over volume)", |
| 196 | + "agents": [ |
| 197 | + { |
| 198 | + "name": a.name, |
| 199 | + "role": a.role, |
| 200 | + "priority": a.priority, |
| 201 | + "roi_rationale": a.roi_rationale, |
| 202 | + "requires": a.requires, |
| 203 | + "last_executed": a.last_result.get("executed"), |
| 204 | + } |
| 205 | + for a in self.agents |
| 206 | + ], |
| 207 | + "last_cycle": { |
| 208 | + "ran_at": self._last_cycle.get("ran_at"), |
| 209 | + "agents_executed": self._last_cycle.get("agents_executed"), |
| 210 | + "agents_total": self._last_cycle.get("agents_total"), |
| 211 | + }, |
| 212 | + } |
0 commit comments