Skip to content

refined-element/le-agent-sdk-python

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

30 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

le-agent-sdk

Discord

PyPI version Tests Python 3.10+ License: MIT

Python SDK for Lightning Enable Agent Service Agreements.

Discover, request, and settle agent-to-agent services over Nostr with L402 Lightning payments.

Installation

pip install le-agent-sdk

Quick Start

Provider: Publish a Service

Register an agent capability on the Nostr network so other agents can discover it.

import asyncio
from le_agent_sdk import AgentManager, AgentCapability, AgentPricing

async def main():
    manager = AgentManager(
        private_key="<your_hex_private_key>",
        relay_urls=["wss://agents.lightningenable.com"],
    )

    cap = AgentCapability(
        service_id="translate-v1",
        categories=["ai", "translation"],
        content="AI translation service. Supports 50+ languages.",
        pricing=[
            AgentPricing(amount=10, unit="sats", model="per-request"),
        ],
        l402_endpoint="https://api.lightningenable.com/l402/proxy/translate",
        hashtags=["translation", "ai"],
    )

    event_id = await manager.publish_capability(cap)
    print(f"Published capability: {event_id}")

    # Listen for incoming service requests
    async for request in manager.listen_requests():
        print(f"Request from {request.pubkey}: {request.content}")

asyncio.run(main())

Requester: Discover and Use Services

Find available services and settle via L402 payments.

import asyncio
from le_agent_sdk import AgentManager

async def main():
    manager = AgentManager(
        private_key="<your_hex_private_key>",
        relay_urls=["wss://agents.lightningenable.com"],
    )

    # Discover translation services
    capabilities = await manager.discover(
        categories=["translation"],
        hashtags=["ai"],
        limit=10,
    )

    for cap in capabilities:
        print(f"[{cap.service_id}] {cap.content[:60]}...")
        if cap.pricing:
            print(f"  Price: {cap.pricing[0].amount} {cap.pricing[0].unit}/{cap.pricing[0].model}")

    # Settle via L402 if endpoint available
    chosen = capabilities[0]
    if chosen.l402_endpoint:
        result = await manager.settle_via_l402(chosen)
        print(f"Result: HTTP {result.status_code}")

asyncio.run(main())

API Reference

Core Classes

Class Description
AgentManager Main entry point. Manages Nostr connections, publishes capabilities, discovers services, and handles L402 settlement.
AgentCapability Defines a service offering with pricing, categories, endpoints, and metadata. Published as Nostr kind 38400 events.
AgentServiceRequest Represents a request for service from one agent to another (kind 38401).
AgentServiceAgreement Bilateral contract between provider and requester (kind 38402).
AgentAttestation Post-completion review of an agent (kind 38403): rating 1-5, review text, optional payment proof. The building block for on-protocol reputation.

Nostr Layer

Class Description
RelayClient WebSocket client for Nostr relay communication. Handles subscriptions and event publishing.
NostrEvent Nostr event construction, serialization, and signing.
TagParser Utilities for parsing and building Nostr event tags.

Payment Layer

Class Description
L402Client HTTP client with automatic L402 challenge-response handling. Wraps l402-requests.
AgentPricing Pricing model (amount, unit, per-request/per-token).

Reputation (AgentManager methods)

Method Description
publish_attestation(subject_pubkey, agreement_id, rating, content="", proof=None) Publish a review of an agent after a completed agreement (kind 38403). rating must be 1-5; proof is an optional hash of the L402 payment preimage. Returns the published AgentAttestation.
get_attestations(pubkey, limit=20, timeout=5.0) Query relays for attestations about an agent. Returns list[AgentAttestation].
get_reputation_score(pubkey, limit=50, timeout=5.0) Average rating (1.0-5.0) computed from attestations, or None if the agent has no attestations yet.

Example: Attest and Check Reputation

import asyncio
from le_agent_sdk import AgentManager

async def main():
    manager = AgentManager(
        private_key="<your_hex_private_key>",
        relay_urls=["wss://agents.lightningenable.com"],
    )

    # After a completed service agreement, publish a review
    attestation = await manager.publish_attestation(
        subject_pubkey="<provider_pubkey>",
        agreement_id="<agreement_event_id>",
        rating=5,
        content="Fast, accurate translation. Would hire again.",
    )
    print(f"Published attestation: {attestation.event_id}")

    # Before hiring an agent, check its track record
    score = await manager.get_reputation_score("<provider_pubkey>")
    if score is None:
        print("No attestations yet")
    else:
        print(f"Reputation: {score:.1f}/5.0")

asyncio.run(main())

Protocol

Agent Service Agreements use four Nostr event kinds:

  • 38400 -- Agent Capability: provider advertises available services
  • 38401 -- Agent Service Request: requester asks for a service
  • 38402 -- Agent Service Agreement: bilateral contract with terms and pricing
  • 38403 -- Agent Attestation: post-completion review (rating 1-5) that builds on-protocol reputation

Settlement happens via L402 (Lightning HTTP 402) through Lightning Enable endpoints.

Related Projects

License

MIT

About

Python SDK for Lightning Enable Agent Service Agreements on Nostr

Resources

License

Stars

0 stars

Watchers

0 watching

Forks

Packages

 
 
 

Contributors

Languages