Build Agent
Infrastructure,
Not Agent Demos.

from maple import Agent, Message, Priority, Config
from maple.core import Result
import asyncio

# Every MAPLE operation returns Result[T, Error].
# Errors are values — no exceptions in your agent graph.

async def main():
    config = Config(agent_id="my_agent", broker_url="memory://localhost")
    agent  = Agent(config)
    await agent.start()

    msg    = Message(message_type="TASK", receiver="worker", priority=Priority.HIGH)
    result = agent.send(msg)

    # ── Pattern 1: Explicit discriminated check (most readable)
    if result.is_ok():
        msg_id = result.unwrap()            # safe: won't raise
        print(f"✓ Delivered: {msg_id}")
    else:
        err  = result.unwrap_err()
        code = err.get('code', 'UNKNOWN')
        txt  = err.get('message', '')
        if err.get('recoverable'):
            print(f"⚠ Recoverable [{code}]: {txt}")
        else:
            print(f"✗ Fatal [{code}]: {txt}")

    # ── Pattern 2: Functional map/map_err chaining
    final = (agent.send(msg)
        .map(lambda mid: f"sent:{mid}")         # transform Ok value
        .map_err(lambda e: {**e, "logged": True}) # enrich Err value
        .unwrap_or("fallback"))                 # safe default on Err

    # ── Pattern 3: Short-circuit with fallback function
    value = agent.send(msg).unwrap_or_else(
        lambda err: attempt_recovery(err)       # called only on Err
    )

    # ── Pattern 4: Force-unwrap — raises ResultError on Err
    #    Use only when you are certain send() cannot fail (e.g. in tests)
    msg_id = agent.send(msg).unwrap()

asyncio.run(main())

from maple import Message, Priority
from maple.resources import ResourceRequest, ResourceRange

# Declare exactly what your workload needs — at the protocol level.
# The scheduler routes to agents that can satisfy the requirements.
# Requests that can't be satisfied return Err — no silent overloads.

heavy_job = Message(
    message_type="ML_INFERENCE",
    receiver="gpu_worker_pool",
    priority=Priority.HIGH,
    payload={
        "model": "llama3-70b",
        "batch_size": 32,
        "resources": ResourceRequest(
            memory=ResourceRange(
                min="16GB", preferred="32GB", max="64GB"
            ),
            compute=ResourceRange(         # CPU cores
                min=8, preferred=16, max=32
            ),
            gpu_memory=ResourceRange(
                min="16GB", preferred="48GB"
            ),
            network_bandwidth=ResourceRange(
                min="1Gbps", preferred="10Gbps"
            ),
            deadline="2025-12-01T18:00:00Z"
        ).to_dict()
    }
)

# If no agent in the pool can satisfy resources,
# send() returns Err instead of silently queuing an unrunnable job.
result = agent.send(heavy_job)

from maple import Agent, Message, Config
from maple.security import SecurityConfig
import asyncio

async def main():
    # 1. Configure with security
    config = Config(
        agent_id="compliance_agent",
        security=SecurityConfig(
            auth_type="JWT",
            credentials={"secret": "your-256-bit-secret"},
            encryption="AES-256-GCM"
        )
    )
    agent = Agent(config)
    await agent.start()

    # 2. Establish a cryptographically-verified channel.
    #    The Link ID proves identity on both ends of the connection.
    link_result = await agent.establish_link(
        target_agent="data_processor",
        security_level="MAXIMUM"
    )

    if link_result.is_ok():
        link_id = link_result.unwrap()   # cryptographic channel token

        # 3. All messages over this link are authenticated
        #    against the Link ID. Forged or expired IDs → Err on delivery.
        sensitive_msg = Message(
            message_type="CONFIDENTIAL_PAYLOAD",
            payload={
                "record_id": "PATIENT-00421",
                "data": "...encrypted payload..."
            }
        ).with_link(link_id)   # MAPLE-exclusive: link-bound message

        result = agent.send(sensitive_msg)
        # Tampering, replay, or MITM breaks channel verification → Err

        if result.is_ok():
            print(f"✓ Secure delivery confirmed: {result.unwrap()}")
        else:
            err = result.unwrap_err()
            print(f"✗ Link verification failed: {err['message']}")
    else:
        print(f"✗ Could not establish link: {link_result.unwrap_err()['message']}")

    await agent.stop()

asyncio.run(main())

from maple.error import CircuitBreaker, CircuitState
from maple.core import Err

# MAPLE's CircuitBreaker is shared infrastructure.
# Configure once, protect your entire agent mesh from cascading failures.

cb = CircuitBreaker(
    failure_threshold=5,       # open after 5 consecutive failures
    recovery_timeout=30.0,     # probe after 30s in OPEN state
    half_open_max_calls=3      # max test calls in HALF_OPEN
)

# State machine: CLOSED → OPEN (threshold hit)
#                        → HALF_OPEN (timeout elapsed)
#                        → CLOSED (probes succeed) | OPEN (probes fail)

async def call_external_service(payload):
    if not cb.should_allow():
        # Circuit is OPEN — fail fast, protect the downstream
        return Err({"code": "CIRCUIT_OPEN", "message": "Service unavailable"})

    result = await external_api.call(payload)

    if result.is_ok():
        cb.record_success()   # may transition HALF_OPEN → CLOSED
    else:
        cb.record_failure()   # may transition CLOSED → OPEN

    return result

# Inspect state for dashboards / alerting
state = cb.get_state()
print(f"State:    {state.state}")           # CLOSED | OPEN | HALF_OPEN
print(f"Failures: {state.failure_count}")
print(f"Since:    {state.last_failure_time}")