Python SDK

pip install 'aether-forge[all] @ git+https://github.com/HeyElsa/aether-forge.git'

from pathlib import Path
from aether_forge import AgentRunner, RunnerConfig
 
runner = AgentRunner(
    Path("./my-agent"),
    config=RunnerConfig(
        interval_seconds=30.0,
        max_ticks=10,
        environment="sandbox",
        auto_approve=True,
        enable_autoresearch=True,
        health_port=8080,
        a2a_port=9001,
        tick_timeout_seconds=120.0,
        circuit_breaker_threshold=5,
    ),
)
 
results = runner.run()
for tick in results:
    print(f"Tick {tick.tick_number}: {tick.session_status} ({tick.steps_executed} steps)")

from aether_forge import Forge
 
project = Forge.generate_fast(
    name="research-agent",
    idea="summarize a webpage and save a short note",
    output="./research-agent",
    planner="heuristic",
)
 
project.validate()
project.eval_pack()
project.run(environment="sandbox", max_ticks=1)

from pathlib import Path
from aether_forge import FastGenerateRequest, generate_fast_artifact_set
 
result = generate_fast_artifact_set(FastGenerateRequest(
    name="my-agent",
    idea="ETH swing trader using momentum signals",
    output_directory=Path("./my-agent"),
    create_wallet=True,
    autonomous=True,
    strategy_file="strategy.md",
    planner_mode="openrouter",
    planner_model="anthropic/claude-sonnet-4",
))
print(f"Generated at {result.output_directory}")
print(f"Wallet: {result.wallet_address}")

from pathlib import Path
 
from aether_forge import AgentRunner, RunnerConfig
 
runner = AgentRunner(
    Path("./my-agent"),
    config=RunnerConfig(environment="sandbox", auto_approve=True),
)
 
result = runner.tick(scenario_inputs={"source": "manual"})
print(f"Status: {result.session_status}")
print(f"Steps: {result.steps_executed}")
print(f"Working set keys: {result.working_set_keys}")

from aether_forge import MemoryQuery, SqliteMemoryStore
 
store = SqliteMemoryStore("./my-agent/memory.db")
 
records = store.read(MemoryQuery(
    memory_type="decision-history",
    scope="session",
    environment="paper",
    limit=20,
))
 
for r in records:
    print(f"[{r.created_at}] {r.summary}")
 
store.close()

from aether_forge.prompting import (
    build_function_call_prompt_from_session,
    estimate_tokens,
    truncate_to_budget,
)
 
# Inside a tick handler...
prompt = build_function_call_prompt_from_session(
    session,
    declared_capability_ids={"cap-market-btc-price", "cap-exchange-order"},
    model="claude-sonnet-4",  # for token budget
)
print(f"Prompt size: {estimate_tokens(prompt)} tokens")

from aether_forge.a2a_client import A2AForgeClient
 
client = A2AForgeClient("http://localhost:9001")
card = client.get_agent_card()
print(f"Connected to: {card.name}")
 
result = client.send_task(
    capability="get-token-price",
    arguments={"token": "ETH"},
)
print(f"Status: {result['status']}")
print(f"Artifacts: {result['artifacts']}")

from aether_forge.defi_safety import (
    simulate_tx, check_slippage, ExposureTracker,
    check_position_health, SwapQuote,
)
 
# Simulate before signing
sim = simulate_tx(
    rpc_url="https://mainnet.base.org",
    from_address="0xE8D0...",
    to_address="0x8335...",
    data="0xa9059cbb...",
)
if not sim.success:
    raise RuntimeError(f"Tx would revert: {sim.revert_reason}")

from aether_forge.x402_client import X402Client, X402Config
from aether_forge.wallet import load_wallet
 
wallet = load_wallet("./my-agent")
 
client = X402Client(
    wallet=wallet,
    config=X402Config(
        max_per_call_usd=0.01,
        max_session_usd=1.00,
        max_daily_usd=5.00,
        chain="base",
        confirmed=True,
    ),
)
 
result = client.get("https://x402-api.heyelsa.ai/api/get_token_price?token=ETH")
print(result)

from pathlib import Path
from aether_forge.attestation import (
    Attestation, verify_self_attestation, determine_trust_tier,
)
 
attestation = Attestation.load(Path("./my-agent/attestation.json"))
ok = verify_self_attestation(attestation)
tier = determine_trust_tier(attestation)
 
print(f"Valid: {ok}, Tier: {tier}")  # e.g., "self-attested"

from aether_forge.a2a_server import A2AServer, build_agent_card
 
# Build card from your agent's spec/manifest
card = build_agent_card(agent_spec, capability_manifest, port=9001)
 
def my_task_handler(message: dict) -> dict:
    capability = message["params"]["message"]["parts"][0]["data"]["capability"]
    # ... your logic ...
    return {"status": "completed", "artifacts": [{"type": "text", "value": "done"}]}
 
server = A2AServer(port=9001, agent_card=card, task_handler=my_task_handler)
server.start()
# ... server runs in background daemon thread ...
server.stop()

from aether_forge.runtime import StepProposal, StepKind, ExecutionResult
from aether_forge.policy import PolicyDecision

from aether_forge import (
    Planner,            # propose_plan(session) -> list[StepProposal]
    ExecutionRouter,    # execute(session, proposal, capability) -> ExecutionResult
    PlanningModel,      # complete(prompt) -> str
    MemoryStore,        # read / write / promote
    DataSource,         # ABC with supports / fetch / subscribe
    Subscription,       # stop(), active
    SecretsProvider,    # get_secret / resolve_api_key
    MarketDataVenue,    # get_price / get_candles
)

from aether_forge import (
    # Planning model implementations
    AnthropicPlanningModel, OpenAICompatiblePlanningModel,
    GeminiPlanningModel, StaticPlanningModel,
    # Data layer concrete sources + types
    HTTPDataSource, X402DataSource, WebSocketDataSource,
    McpDataSource, MockDataSource,
    DataResult, DataRouter, DataSourceCost,
    # Memory
    InMemoryMemoryStore, SqliteMemoryStore,
    # Execution
    MockCryptoExecutionRouter,
)

import aether_forge
 
assert aether_forge.API_STABILITY["FastGenerateRequest"] == "stable"
assert aether_forge.API_STABILITY["X402Client"] == "experimental"