ucm-sdk 0.2.0

Python SDK for UCM — Agent-Native API Marketplace

UCM Python SDK

Python SDK for the UCM Agent Commerce Protocol.

Installation

pip install ucm-sdk

For development:

pip install ucm-sdk[dev]

Quick Start

API Key mode (simplest)

import asyncio
from ucm import UCMClient

async def main():
    # Register and get an API key
    async with UCMClient('https://registry.ucm.ai', api_key='ucm_key_...') as client:
        # Discover services
        result = await client.discover('I need web search API')
        for svc in result['services']:
            print(f"{svc['name']} - {svc['price']}")

        # Buy a service
        receipt = await client.buy('acme/web-search')
        print(receipt['access_token'])

        # Check balance
        balance = await client.balance()
        print(balance)

asyncio.run(main())

Ed25519 mode (provider operations)

import asyncio
from ucm import UCMClient, generate_keypair

async def main():
    private_key, public_key = generate_keypair()

    async with UCMClient('https://registry.ucm.ai', private_key, public_key) as client:
        # Publish a service (requires Ed25519)
        result = await client.publish({
            'acp_version': '0.1',
            'service_id': 'myorg/my-api',
            'name': 'My API',
            'description': 'Does something useful',
            'provider': {'wallet': public_key, 'name': 'My Org'},
            'base_url': 'https://api.example.com',
            'endpoints': [{'path': '/v1/query', 'method': 'POST'}],
            'tags': ['search'],
            'pricing': {
                'model': 'per-call',
                'price': '0.01',
                'currency': 'USDT',
                'unit': 'call',
            },
            'sla': {'uptime': '99.9%', 'latency_p95': '500ms'},
        })
        print(result['service_id'], result['status'])

asyncio.run(main())

Self-registration flow

import asyncio
from ucm import UCMClient

async def main():
    # Start without any auth — register to get an API key
    async with UCMClient('https://registry.ucm.ai', api_key='placeholder') as client:
        pass  # placeholder needed for constructor

    # Better: use register which auto-sets the api_key
    client = UCMClient('https://registry.ucm.ai', api_key='_')
    reg = await client.register('my-agent', description='A helpful agent')
    # client.api_key is now automatically set
    print(f"Agent ID: {reg['agent_id']}")
    print(f"API Key: {reg['api_key']}")

    # Now use authenticated endpoints
    balance = await client.balance()
    print(balance)
    await client.close()

asyncio.run(main())

Error handling

from ucm import UCMClient, ACPError, WalletRequiredError, BudgetError

async def safe_buy(client: UCMClient, service_id: str):
    try:
        return await client.buy(service_id)
    except WalletRequiredError:
        print("Need to bind a wallet first!")
    except BudgetError as e:
        print(f"Budget issue: {e.message}")
    except ACPError as e:
        print(f"ACP error [{e.code}]: {e.message} (HTTP {e.status_code})")

API Reference

generate_keypair() -> tuple[str, str]

Generate a new Ed25519 keypair. Returns (private_key_hex, public_key_hex).

UCMClient

Async client for interacting with the UCM registry.

Constructor

UCMClient(
    registry_url: str,
    private_key: str | None = None,
    public_key: str | None = None,
    api_key: str | None = None,
    timeout: float = 30.0,
)

At least one of (private_key, public_key) or api_key must be provided.

Public endpoints (no auth)

Method	Description
discover(need, tags?, limit?)	Search services by natural language
service_info(service_id)	Get service details
verify_token(token)	Verify an access token
deposit_info()	Get Solana escrow information
get_onboarding_md()	Fetch agent onboarding guide

Agent management

Method	Description
register(name, description?)	Self-register, get API key
bind_wallet(wallet, signature)	Bind Solana wallet
recover(wallet, timestamp, signature)	Recover account via wallet

Authenticated endpoints (Ed25519 or API key)

Method	Description
buy(service_id, plan?)	Purchase service access
balance()	Check budget and usage
history(limit?)	Get transaction history

Ed25519-only endpoints (provider/operator)

Method	Description
publish(descriptor)	List a new service
update_service(service_id, update)	Update a service
delete_service(service_id)	Delist a service
authorize(agent, rules, operator?)	Create budget authorization

Exceptions

All ACP errors are mapped to typed exceptions:

Exception	ACP Error Codes
ACPError	Base class for all errors
InvalidRequestError	INVALID_REQUEST, INVALID_DESCRIPTOR
AuthenticationError	INVALID_SIGNATURE, NONCE_REUSED, TIMESTAMP_EXPIRED
BudgetError	BUDGET_RULE_VIOLATED, INSUFFICIENT_BUDGET, AUTHORIZATION_EXPIRED, APPROVAL_REQUIRED
NotFoundError	SERVICE_NOT_FOUND, AGENT_NOT_FOUND
ConflictError	SERVICE_ALREADY_EXISTS, WALLET_ALREADY_BOUND, WALLET_IN_USE
WalletRequiredError	WALLET_REQUIRED
RateLimitError	RATE_LIMITED

Each exception has .code, .message, .status_code, and .detail attributes.

Authentication

The SDK supports two authentication modes:

1. Ed25519 signatures — Required for provider operations (publish, authorize, delete). Headers: X-ACP-Pubkey, X-ACP-Timestamp, X-ACP-Nonce, X-ACP-Signature.

2. API Key (Bearer token) — Simpler auth for agent operations (buy, balance, history). Header: Authorization: Bearer ucm_key_....

When both are configured, Ed25519 is preferred. API key is used as fallback.

License

MIT