From ed87ae4c9345ddfcf2ea6ca85438a7acb3a5719e Mon Sep 17 00:00:00 2001 From: eruizgar91 <3496824+eruizgar91@users.noreply.github.com> Date: Fri, 13 Feb 2026 14:20:06 +0000 Subject: [PATCH] docs: update Python SDK documentation for v1.2.1 - Updated from nevermined-io/payments-py@fb4b2d7df97ccc150df05a04cecd91cd59e9f36f - SDK version: v1.2.1 - Target branch: main - Generated documentation from tagged release - Converted to Mintlify MDX format --- docs/api-reference/python/a2a-module.mdx | 362 +++++++++--------- .../python/validation-module.mdx | 48 ++- docs/api-reference/python/x402-module.mdx | 54 ++- 3 files changed, 249 insertions(+), 215 deletions(-) diff --git a/docs/api-reference/python/a2a-module.mdx b/docs/api-reference/python/a2a-module.mdx index 0ee73d5..29bf945 100644 --- a/docs/api-reference/python/a2a-module.mdx +++ b/docs/api-reference/python/a2a-module.mdx @@ -14,84 +14,58 @@ A2A (Agent-to-Agent) is a protocol that enables AI agents to communicate with ea - Automatically verify x402 tokens on incoming requests - Handle credit redemption for agent tasks -## A2A Integration API - -Access the A2A integration through `payments.a2a`: - -```python -from payments_py import Payments, PaymentOptions - -payments = Payments.get_instance( - PaymentOptions(nvm_api_key="nvm:your-key", environment="sandbox") -) - -# A2A integration is available as: -a2a = payments.a2a -``` - -## Starting an A2A Server +## Building Agent Cards -### Basic Server Setup +### With a Single Plan ```python -from payments_py import Payments, PaymentOptions from payments_py.a2a.agent_card import build_payment_agent_card -from a2a.server.agent_execution import AgentExecutor -# Initialize payments -payments = Payments.get_instance( - PaymentOptions(nvm_api_key="nvm:your-key", environment="sandbox") -) - -# Build agent card with payment extension agent_card = build_payment_agent_card( - name="My AI Agent", - description="An AI agent with payment support", - agent_id="your-agent-id", - plan_id="your-plan-id", - url="https://your-agent.com" + base_card={ + "name": "Code Review Agent", + "description": "Automated code review powered by AI", + "url": "https://localhost:8080", + "version": "1.0.0", + "capabilities": { + "streaming": False, + "pushNotifications": False, + }, + }, + payment_metadata={ + "paymentType": "fixed", + "credits": 1, + "agentId": "your-agent-id", + "planId": "your-plan-id", + "costDescription": "1 credit per review", + }, ) - -# Create your agent executor (implements A2A AgentExecutor interface) -class MyAgentExecutor(AgentExecutor): - async def execute(self, task, context): - # Your agent logic here - return {"result": "Task completed"} - -executor = MyAgentExecutor() - -# Start the A2A server -result = payments.a2a["start"]( - agent_card=agent_card, - executor=executor, - port=8080 -) - -# Run the server -result.server.run() ``` -## Building Agent Cards +### With Multiple Plans -### With Payment Extension +When your agent supports multiple plans (e.g. a basic and a premium tier), use `planIds` instead of `planId`: ```python -from payments_py.a2a.agent_card import build_payment_agent_card - agent_card = build_payment_agent_card( - name="Code Review Agent", - description="Automated code review powered by AI", - agent_id="did:nvm:abc123", - plan_id="plan-456", - url="https://code-review.example.com", - version="1.0.0", - capabilities={ - "streaming": True, - "pushNotifications": True - } + base_card={ + "name": "Code Review Agent", + "url": "https://localhost:8080", + "version": "1.0.0", + "capabilities": {}, + }, + payment_metadata={ + "paymentType": "dynamic", + "credits": 5, + "agentId": "your-agent-id", + "planIds": ["plan-basic", "plan-premium"], + "costDescription": "1-5 credits depending on review depth", + }, ) ``` +> **Note:** Provide either `planId` or `planIds`, not both. `planIds` must be a non-empty list. + ### Agent Card Structure The agent card includes a payment extension: @@ -99,18 +73,18 @@ The agent card includes a payment extension: ```json { "name": "Code Review Agent", - "description": "Automated code review powered by AI", - "url": "https://code-review.example.com", + "url": "https://localhost:8080", "version": "1.0.0", "capabilities": { - "streaming": true, - "pushNotifications": true, "extensions": [ { "uri": "urn:nevermined:payment", "params": { - "agentId": "did:nvm:abc123", - "planId": "plan-456" + "paymentType": "dynamic", + "credits": 5, + "agentId": "your-agent-id", + "planIds": ["plan-basic", "plan-premium"], + "costDescription": "1-5 credits depending on review depth" } } ] @@ -118,6 +92,93 @@ The agent card includes a payment extension: } ``` +## Using the `@a2a_requires_payment` Decorator + +The simplest way to create a payment-protected A2A agent: + +```python +from payments_py import Payments +from payments_py.common.types import PaymentOptions +from payments_py.a2a import AgentResponse, a2a_requires_payment, build_payment_agent_card + +payments = Payments.get_instance( + PaymentOptions(nvm_api_key="nvm:your-key", environment="sandbox") +) + +agent_card = build_payment_agent_card( + base_card={ + "name": "My Agent", + "url": "http://localhost:8080", + "version": "1.0.0", + "capabilities": {}, + }, + payment_metadata={ + "paymentType": "dynamic", + "credits": 3, + "agentId": "your-agent-id", + "planIds": ["plan-1", "plan-2"], + }, +) + +@a2a_requires_payment( + payments=payments, + agent_card=agent_card, + default_credits=1, +) +async def my_agent(context) -> AgentResponse: + text = context.get_user_input() + return AgentResponse(text=f"Echo: {text}", credits_used=1) + +# Start serving (blocking) +my_agent.serve(port=8080) +``` + +The decorator handles: + +- Payment middleware (verify/settle) automatically +- Publishing task status events with `creditsUsed` metadata +- Credit burning on task completion + +### `AgentResponse` + +| Field | Type | Description | +|-------|------|-------------| +| `text` | `str` | The agent's text response | +| `credits_used` | `int \| None` | Credits consumed (falls back to `default_credits`) | +| `metadata` | `dict \| None` | Extra metadata for the final event | + +## Starting an A2A Server (Advanced) + +For more control, use `PaymentsA2AServer.start()` directly: + +```python +from payments_py.a2a.server import PaymentsA2AServer +from a2a.server.agent_execution import AgentExecutor +from a2a.server.events.event_queue import EventQueue + +class MyExecutor(AgentExecutor): + async def execute(self, context, event_queue: EventQueue): + # Your agent logic — publish events to event_queue + ... + + async def cancel(self, context, event_queue: EventQueue): + ... + +result = PaymentsA2AServer.start( + agent_card=agent_card, + executor=MyExecutor(), + payments_service=payments, + port=8080, + base_path="/", + expose_agent_card=True, + async_execution=False, +) + +# Run the server +import asyncio +asyncio.run(result.server.serve()) +``` + ## Server Configuration Options | Option | Type | Required | Description | @@ -134,136 +195,79 @@ The agent card includes a payment extension: ## Request Validation -The A2A server automatically validates payments: +The A2A server automatically validates payments on every POST request: 1. Extracts Bearer token from Authorization header -2. Verifies permissions against the agent's plan -3. Rejects requests with 402 if validation fails +2. Reads `planId` or `planIds` from the agent card's payment extension +3. Verifies permissions via `build_payment_required_for_plans()` +4. Rejects requests with 402 if validation fails, including a base64-encoded `payment-required` header -```python -# Requests must include: -# Authorization: Bearer +When multiple plans are configured, the 402 response includes all plans in `accepts[]`, allowing the client to choose which plan to purchase. -# Invalid requests receive: -# HTTP 402 Payment Required -# {"error": {"code": -32001, "message": "Validation error: ..."}} ``` +HTTP/1.1 402 Payment Required +payment-required: eyJ4NDAy... (base64-encoded X402PaymentRequired) -## Complete Example - -```python -import asyncio -from payments_py import Payments, PaymentOptions -from payments_py.a2a.agent_card import build_payment_agent_card -from a2a.server.agent_execution import AgentExecutor -from a2a.types import Task, TaskResult, Message - -# Initialize payments -payments = Payments.get_instance( - PaymentOptions(nvm_api_key="nvm:your-key", environment="sandbox") -) +{"error": {"code": -32001, "message": "Missing bearer token."}} +``` -# Define your agent executor -class CodeReviewExecutor(AgentExecutor): - """Agent that performs code reviews.""" - - async def execute(self, task: Task, context: dict) -> TaskResult: - # Extract code from task - messages = task.get("messages", []) - code = "" - for msg in messages: - if msg.get("role") == "user": - parts = msg.get("parts", []) - for part in parts: - if part.get("type") == "text": - code = part.get("text", "") - break - - # Perform code review (your logic here) - review_result = await self.review_code(code) - - return TaskResult( - status="completed", - messages=[ - Message( - role="assistant", - parts=[{"type": "text", "text": review_result}] - ) - ] - ) - - async def review_code(self, code: str) -> str: - # Your code review logic - return f"Code review completed. Found 3 suggestions for improvement." - -# Build agent card -agent_card = build_payment_agent_card( - name="Code Review Agent", - description="AI-powered code review service", - agent_id="did:nvm:code-review-agent", - plan_id="code-review-plan-id", - url="https://localhost:8080", - version="1.0.0" -) +## Client Usage -# Start server -async def main(): - executor = CodeReviewExecutor() +### Discovering Plans from the Agent Card - result = payments.a2a["start"]( - agent_card=agent_card, - executor=executor, - port=8080, - expose_agent_card=True, - async_execution=True - ) +Consumers can fetch the agent card to discover available plans: - print(f"A2A Server started on port 8080") - print(f"Agent card: http://localhost:8080/.well-known/agent.json") +```python +import httpx - # Run server - await result.server.serve() +async with httpx.AsyncClient() as client: + resp = await client.get("http://agent-url/.well-known/agent.json") + card = resp.json() -asyncio.run(main()) +extensions = card["capabilities"]["extensions"] +payment_ext = next(e for e in extensions if e["uri"] == "urn:nevermined:payment") +plan_ids = payment_ext["params"].get("planIds") or [payment_ext["params"]["planId"]] +agent_id = payment_ext["params"]["agentId"] ``` -## Client Usage - -Clients interact with the A2A server using the standard A2A protocol: +### Ordering a Plan and Sending Messages ```python from payments_py import Payments, PaymentOptions -# Initialize as subscriber payments = Payments.get_instance( PaymentOptions(nvm_api_key="nvm:subscriber-key", environment="sandbox") ) -# Get access token -token_result = payments.x402.get_x402_access_token( - plan_id="code-review-plan-id", - agent_id="did:nvm:code-review-agent" -) -access_token = token_result['accessToken'] +# Order (purchase) the plan +payments.plans.order_plan(plan_ids[0]) -# Get A2A client -client = payments.a2a["get_client"]( - agent_url="https://localhost:8080", - access_token=access_token +# Get x402 access token +token_resp = payments.x402.get_x402_access_token( + plan_id=plan_ids[0], + agent_id=agent_id, ) - -# Send a task -response = await client.send_task({ - "messages": [{ - "role": "user", - "parts": [{ - "type": "text", - "text": "def add(a, b):\n return a + b" - }] - }] -}) - -print(f"Review result: {response}") +access_token = token_resp["accessToken"] + +# Send A2A JSON-RPC message +async with httpx.AsyncClient() as client: + resp = await client.post( + "http://agent-url/", + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "message/send", + "params": { + "message": { + "messageId": "msg-1", + "role": "user", + "parts": [{"kind": "text", "text": "Review this code"}], + } + }, + }, + headers={"Authorization": f"Bearer {access_token}"}, + ) + result = resp.json() ``` ## Hooks @@ -280,14 +284,15 @@ async def after_request(method, response, request): async def on_error(method, error, request): print(f"Request failed: {method} - {error}") -result = payments.a2a["start"]( +result = PaymentsA2AServer.start( agent_card=agent_card, executor=executor, + payments_service=payments, hooks={ "beforeRequest": before_request, "afterRequest": after_request, - "onError": on_error - } + "onError": on_error, + }, ) ``` @@ -295,9 +300,10 @@ result = payments.a2a["start"]( | Error Code | HTTP Status | Description | |------------|-------------|-------------| -| -32001 | 401 | Missing Bearer token | +| -32001 | 402 | Missing Bearer token | | -32001 | 402 | Payment validation failed | | -32001 | 402 | Agent ID missing from card | +| -32001 | 402 | Plan ID missing from card | ## Next Steps diff --git a/docs/api-reference/python/validation-module.mdx b/docs/api-reference/python/validation-module.mdx index 75c309f..2d65e3d 100644 --- a/docs/api-reference/python/validation-module.mdx +++ b/docs/api-reference/python/validation-module.mdx @@ -73,9 +73,9 @@ verification = payments.facilitator.verify_permissions( if verification.is_valid: print("Request is valid!") - print(f"Subscriber: {verification.subscriber_address}") + print(f"Payer: {verification.payer}") else: - print(f"Invalid request: {verification.error}") + print(f"Invalid request: {verification.invalid_reason}") ``` ### Settle Permissions @@ -91,7 +91,7 @@ settlement = payments.facilitator.settle_permissions( if settlement.success: print(f"Credits redeemed: {settlement.credits_redeemed}") - print(f"Transaction: {settlement.tx_hash}") + print(f"Transaction: {settlement.transaction}") ``` ## Complete Example: Flask Agent @@ -138,7 +138,7 @@ def create_task(): if not verification.is_valid: return jsonify({ 'error': 'Payment verification failed', - 'details': verification.error + 'details': verification.invalid_reason }), 402 # 4. Process the request @@ -240,17 +240,26 @@ For FastAPI applications, use the built-in x402 middleware: ```python from fastapi import FastAPI -from payments_py.x402.fastapi import X402Middleware +from payments_py import Payments, PaymentOptions +from payments_py.x402.fastapi import PaymentMiddleware app = FastAPI() +# Initialize payments +payments = Payments.get_instance( + PaymentOptions(nvm_api_key="nvm:agent-key", environment="sandbox") +) + # Add x402 middleware app.add_middleware( - X402Middleware, - nvm_api_key="nvm:agent-key", - environment="sandbox", - agent_id="your-agent-id", - plan_id="your-plan-id" + PaymentMiddleware, + payments=payments, + routes={ + "/api/tasks": { + "plan_id": "your-plan-id", + "agent_id": "your-agent-id", + } + } ) @app.post("/api/tasks") @@ -262,26 +271,27 @@ async def create_task(body: dict): ## Verification Response -The `verify_permissions` method returns: +The `verify_permissions` method returns a `VerifyResponse`: | Field | Type | Description | |-------|------|-------------| | `is_valid` | `bool` | Whether the request is authorized | -| `subscriber_address` | `str` | Subscriber's wallet address | -| `plan_id` | `str` | Plan being used | -| `balance` | `int` | Current credit balance | -| `error` | `str` | Error message if invalid | +| `invalid_reason` | `str` | Reason for invalidity (if `is_valid` is false) | +| `payer` | `str` | Payer's wallet address | +| `agent_request_id` | `str` | Agent request ID for observability tracking | ## Settlement Response -The `settle_permissions` method returns: +The `settle_permissions` method returns a `SettleResponse`: | Field | Type | Description | |-------|------|-------------| | `success` | `bool` | Whether settlement succeeded | -| `credits_redeemed` | `int` | Number of credits burned | -| `tx_hash` | `str` | Blockchain transaction hash | -| `remaining_balance` | `int` | Credits remaining | +| `error_reason` | `str` | Reason for settlement failure (if `success` is false) | +| `payer` | `str` | Payer's wallet address | +| `transaction` | `str` | Blockchain transaction hash | +| `credits_redeemed` | `str` | Number of credits burned | +| `remaining_balance` | `str` | Credits remaining | ## Best Practices diff --git a/docs/api-reference/python/x402-module.mdx b/docs/api-reference/python/x402-module.mdx index 9947e2f..0b12f22 100644 --- a/docs/api-reference/python/x402-module.mdx +++ b/docs/api-reference/python/x402-module.mdx @@ -106,10 +106,9 @@ verification = payments.facilitator.verify_permissions( ) if verification.is_valid: - print(f"Valid! Subscriber: {verification.subscriber_address}") - print(f"Balance: {verification.balance}") + print(f"Valid! Payer: {verification.payer}") else: - print(f"Invalid: {verification.error}") + print(f"Invalid: {verification.invalid_reason}") ``` ### Verification Response @@ -117,10 +116,9 @@ else: | Field | Type | Description | |-------|------|-------------| | `is_valid` | `bool` | Whether verification passed | -| `subscriber_address` | `str` | Subscriber's wallet address | -| `plan_id` | `str` | Plan being used | -| `balance` | `int` | Current credit balance | -| `error` | `str` | Error message if invalid | +| `invalid_reason` | `str` | Reason for invalidity (if `is_valid` is false) | +| `payer` | `str` | Payer's wallet address | +| `agent_request_id` | `str` | Agent request ID for observability tracking | ## Settle Payment Permissions @@ -137,10 +135,10 @@ settlement = payments.facilitator.settle_permissions( if settlement.success: print(f"Settled! Credits burned: {settlement.credits_redeemed}") - print(f"Transaction: {settlement.tx_hash}") + print(f"Transaction: {settlement.transaction}") print(f"Remaining: {settlement.remaining_balance}") else: - print(f"Settlement failed: {settlement.error}") + print(f"Settlement failed: {settlement.error_reason}") ``` ### Settlement Response @@ -148,44 +146,64 @@ else: | Field | Type | Description | |-------|------|-------------| | `success` | `bool` | Whether settlement succeeded | -| `credits_redeemed` | `int` | Credits that were burned | -| `tx_hash` | `str` | Blockchain transaction hash | -| `remaining_balance` | `int` | Credits remaining | +| `error_reason` | `str` | Reason for failure (if `success` is false) | +| `payer` | `str` | Payer's wallet address | +| `transaction` | `str` | Blockchain transaction hash | +| `credits_redeemed` | `str` | Credits that were burned | +| `remaining_balance` | `str` | Credits remaining | ## Payment Required Object The `X402PaymentRequired` object specifies what payment is required: ```python -from payments_py.x402.types import X402PaymentRequired, X402Scheme +from payments_py.x402.types import X402PaymentRequired, X402Resource, X402Scheme, X402SchemeExtra payment_required = X402PaymentRequired( x402_version=2, + resource=X402Resource( + url="https://your-api.com/endpoint", + description="Protected endpoint" # Optional + ), accepts=[ X402Scheme( scheme="nvm:erc4337", network="eip155:84532", # Base Sepolia - plan_id="your-plan-id" + plan_id="your-plan-id", + extra=X402SchemeExtra( + http_verb="POST", # HTTP method goes in extra, not resource + agent_id="agent-123" # Optional + ) ) ], extensions={} ) ``` -### Using the Helper +### Using the Helpers ```python -from payments_py.x402.helpers import build_payment_required +from payments_py.x402.helpers import build_payment_required, build_payment_required_for_plans -# Simpler way to build payment required +# Single plan payment_required = build_payment_required( plan_id="your-plan-id", endpoint="https://api.example.com/tasks", agent_id="agent-123", http_verb="POST" ) + +# Multiple plans — creates one entry per plan in accepts[] +payment_required = build_payment_required_for_plans( + plan_ids=["plan-basic", "plan-premium"], + endpoint="https://api.example.com/tasks", + agent_id="agent-123", + http_verb="POST" +) ``` +For a single plan, `build_payment_required_for_plans` delegates to `build_payment_required` internally. + ## Complete Workflow Example ```python @@ -228,7 +246,7 @@ def process_request(): if not verification.is_valid: return jsonify({ 'error': 'Payment required', - 'details': verification.error, + 'details': verification.invalid_reason, 'paymentRequired': payment_required.model_dump() }), 402