diff --git a/python/coinbase-agentkit/changelog.d/add-blockrun-action-provider.feature.md b/python/coinbase-agentkit/changelog.d/add-blockrun-action-provider.feature.md new file mode 100644 index 000000000..26650d332 --- /dev/null +++ b/python/coinbase-agentkit/changelog.d/add-blockrun-action-provider.feature.md @@ -0,0 +1 @@ +Add BlockRun action provider for pay-per-request LLM access via x402 micropayments. BlockRun enables agents to access multiple LLM providers (OpenAI, Anthropic, Google, DeepSeek) with USDC payments on Base chain. Install with `pip install coinbase-agentkit[blockrun]`. diff --git a/python/coinbase-agentkit/coinbase_agentkit/__init__.py b/python/coinbase-agentkit/coinbase_agentkit/__init__.py index 000d587db..27b41f1a6 100644 --- a/python/coinbase-agentkit/coinbase_agentkit/__init__.py +++ b/python/coinbase-agentkit/coinbase_agentkit/__init__.py @@ -6,6 +6,7 @@ ActionProvider, aave_action_provider, basename_action_provider, + blockrun_action_provider, cdp_api_action_provider, cdp_evm_wallet_action_provider, cdp_smart_wallet_action_provider, @@ -47,6 +48,7 @@ "ActionProvider", "create_action", "basename_action_provider", + "blockrun_action_provider", "WalletProvider", "CdpEvmWalletProvider", "CdpEvmWalletProviderConfig", diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/__init__.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/__init__.py index 5f310605c..ae711b0b8 100644 --- a/python/coinbase-agentkit/coinbase_agentkit/action_providers/__init__.py +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/__init__.py @@ -7,6 +7,10 @@ BasenameActionProvider, basename_action_provider, ) +from .blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + blockrun_action_provider, +) from .cdp.cdp_api_action_provider import CdpApiActionProvider, cdp_api_action_provider from .cdp.cdp_evm_wallet_action_provider import ( CdpEvmWalletActionProvider, @@ -46,6 +50,8 @@ "aave_action_provider", "BasenameActionProvider", "basename_action_provider", + "BlockrunActionProvider", + "blockrun_action_provider", "CdpApiActionProvider", "cdp_api_action_provider", "CdpEvmWalletActionProvider", diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/README.md b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/README.md new file mode 100644 index 000000000..bbe0917d2 --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/README.md @@ -0,0 +1,198 @@ +# BlockRun Action Provider + +Access multiple frontier LLMs (GPT-4o, Claude, Gemini, DeepSeek) with pay-per-request USDC micropayments on Base chain. + +## How It Works + +BlockRun uses the **x402 protocol** - an HTTP-native payment standard built by Coinbase: + +``` +┌─────────────┐ 1. Request LLM ┌─────────────┐ +│ AgentKit │ ──────────────────────► │ BlockRun │ +│ + Wallet │ │ API │ +│ Provider │ ◄────────────────────── │ │ +└─────────────┘ 2. 402 + price info └─────────────┘ + │ │ + │ 3. Sign payment │ + │ (via wallet provider) │ + ▼ │ +┌─────────────┐ │ +│ x402 │ 4. Retry with signed │ +│ Library │ payment header │ +│ │ ──────────────────────────────►│ +└─────────────┘ │ + │ │ + │ 5. Verify & settle via │ + │ CDP Facilitator │ + │◄──────────────────────────────────────┘ + │ + ▼ + 6. LLM Response +``` + +**Key point:** Your private keys are **never exposed**. The wallet provider handles signing through its secure infrastructure (CDP server-side signing, Viem, etc.). + +## Installation + +BlockRun uses AgentKit's built-in x402 support - **no extra dependencies needed**: + +```bash +pip install coinbase-agentkit +``` + +## Usage + +```python +from coinbase_agentkit import AgentKit, AgentKitConfig, blockrun_action_provider +from coinbase_agentkit.wallet_providers import CdpEvmWalletProvider, CdpEvmWalletProviderConfig + +# Setup wallet provider (works with ANY AgentKit wallet provider) +wallet_provider = CdpEvmWalletProvider(CdpEvmWalletProviderConfig( + api_key_id="your-cdp-api-key-id", + api_key_secret="your-cdp-api-key-secret", + wallet_secret="your-wallet-secret", + network_id="base-mainnet", +)) + +# Create AgentKit with BlockRun +agentkit = AgentKit(AgentKitConfig( + wallet_provider=wallet_provider, + action_providers=[blockrun_action_provider()], +)) + +# Make an LLM request - payment handled automatically! +result = agentkit.execute_action( + "chat_completion", + { + "prompt": "What is the capital of France?", + "model": "openai/gpt-4o-mini", + } +) +print(result) +``` + +## Wallet Provider Compatibility + +BlockRun works with **any AgentKit EVM wallet provider**: + +| Provider | Signing Location | Use Case | +|----------|------------------|----------| +| `CdpEvmWalletProvider` | Coinbase servers | Production (recommended) | +| `CdpSmartWalletProvider` | Coinbase servers | Smart contract wallets | +| `EthAccountWalletProvider` | Local | Testing & development | +| `ViemWalletProvider` | Configurable | Viem integration | +| `PrivyEvmWalletProvider` | Privy servers | Privy apps | + +**No private key environment variables needed!** The wallet provider handles all signing securely. + +## Available Actions + +### `chat_completion` + +Send a chat completion request to an LLM. + +| Parameter | Type | Required | Default | Description | +|-----------|------|----------|---------|-------------| +| `prompt` | string | Yes | - | The user message to send | +| `model` | string | No | `openai/gpt-4o-mini` | Model to use | +| `system_prompt` | string | No | - | System prompt for context | +| `max_tokens` | integer | No | 1024 | Maximum tokens in response | +| `temperature` | float | No | 0.7 | Sampling temperature (0-2) | + +**Available Models:** + +| Model ID | Provider | Description | +|----------|----------|-------------| +| `openai/gpt-4o` | OpenAI | Most capable GPT-4 with vision | +| `openai/gpt-4o-mini` | OpenAI | Fast and cost-effective | +| `anthropic/claude-sonnet-4` | Anthropic | Balanced for most tasks | +| `google/gemini-2.0-flash` | Google | Fast multimodal model | +| `deepseek/deepseek-chat` | DeepSeek | General-purpose chat | + +**Example:** +```python +result = agentkit.execute_action("chat_completion", { + "model": "anthropic/claude-sonnet-4", + "prompt": "Explain quantum computing in simple terms", + "system_prompt": "You are a physics teacher", + "max_tokens": 500, +}) +``` + +### `get_usdc_balance` + +Check your wallet's USDC balance on Base before making requests. + +```python +result = agentkit.execute_action("get_usdc_balance", {}) +``` + +**Response:** +```json +{ + "success": true, + "balance": 5.123456, + "formatted_balance": "5.123456 USDC", + "wallet_address": "0x...", + "usdc_contract": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", + "network": "base-mainnet" +} +``` + +If balance is low (< 0.10 USDC), a warning and funding suggestions are included. + +### `list_models` + +List all available models with descriptions. No parameters required. + +```python +result = agentkit.execute_action("list_models", {}) +``` + +## Network Support + +| Network | ID | Status | +|---------|----|----| +| Base Mainnet | `base-mainnet` | Production | +| Base Sepolia | `base-sepolia` | Testing | + +Ensure your wallet has USDC on the appropriate network. + +## Payment Details + +- **Currency:** USDC on Base +- **Protocol:** x402 v2 +- **Settlement:** Coinbase CDP Facilitator +- **Pricing:** Pay-per-token (~$0.001-$0.01 per typical request) + +## Response Format + +**Success:** +```json +{ + "success": true, + "model": "openai/gpt-4o-mini", + "response": "The capital of France is Paris.", + "usage": { + "prompt_tokens": 10, + "completion_tokens": 8, + "total_tokens": 18 + }, + "payment": "Paid via x402 micropayment on Base" +} +``` + +**Error:** +```json +{ + "error": true, + "message": "BlockRun chat completion failed: ...", + "suggestion": "Ensure your wallet has sufficient USDC on Base." +} +``` + +## Links + +- [BlockRun Documentation](https://blockrun.ai/docs) +- [x402 Protocol](https://www.x402.org/) +- [Coinbase Developer Platform](https://docs.cdp.coinbase.com/) diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/__init__.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/__init__.py new file mode 100644 index 000000000..9ba98ccbe --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/__init__.py @@ -0,0 +1,5 @@ +"""BlockRun action provider for pay-per-request LLM access via x402 micropayments.""" + +from .blockrun_action_provider import BlockrunActionProvider, blockrun_action_provider + +__all__ = ["BlockrunActionProvider", "blockrun_action_provider"] diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/blockrun_action_provider.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/blockrun_action_provider.py new file mode 100644 index 000000000..2f391d224 --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/blockrun_action_provider.py @@ -0,0 +1,423 @@ +"""BlockRun action provider for pay-per-request LLM access via x402 micropayments. + +BlockRun provides access to multiple LLM providers (OpenAI, Anthropic, Google, DeepSeek) +with pay-per-request USDC micropayments on Base chain using the x402 protocol. +No API keys needed - payments are signed using AgentKit's wallet provider. +""" + +import json +from typing import Any + +from x402.clients.requests import x402_requests + +from ...network import Network +from ...wallet_providers.evm_wallet_provider import EvmWalletProvider +from ..action_decorator import create_action +from ..action_provider import ActionProvider +from .schemas import ChatCompletionSchema, GetUsdcBalanceSchema, ListModelsSchema + +SUPPORTED_NETWORKS = ["base-mainnet", "base-sepolia"] + +BLOCKRUN_API_URL = "https://blockrun.ai/api/v1" + +# USDC contract addresses on Base +USDC_ADDRESSES = { + "base-mainnet": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", + "base-sepolia": "0x036CbD53842c5426634e7929541eC2318f3dCF7e", +} + +# Minimal ERC20 ABI for balance checking +ERC20_BALANCE_ABI = [ + { + "type": "function", + "name": "balanceOf", + "stateMutability": "view", + "inputs": [{"name": "account", "type": "address"}], + "outputs": [{"type": "uint256"}], + }, + { + "type": "function", + "name": "decimals", + "stateMutability": "view", + "inputs": [], + "outputs": [{"type": "uint8"}], + }, +] + +AVAILABLE_MODELS = { + "openai/gpt-4o": { + "name": "GPT-4o", + "provider": "OpenAI", + "description": "Most capable GPT-4 model with vision capabilities", + }, + "openai/gpt-4o-mini": { + "name": "GPT-4o Mini", + "provider": "OpenAI", + "description": "Fast and cost-effective GPT-4 model", + }, + "anthropic/claude-sonnet-4": { + "name": "Claude Sonnet 4", + "provider": "Anthropic", + "description": "Anthropic's balanced model for most tasks", + }, + "google/gemini-2.0-flash": { + "name": "Gemini 2.0 Flash", + "provider": "Google", + "description": "Google's fast multimodal model", + }, + "deepseek/deepseek-chat": { + "name": "DeepSeek Chat", + "provider": "DeepSeek", + "description": "DeepSeek's general-purpose chat model", + }, +} + + +class BlockrunActionProvider(ActionProvider[EvmWalletProvider]): + """Action provider for BlockRun LLM services via x402 micropayments. + + BlockRun enables AI agents to access multiple LLM providers using pay-per-request + USDC micropayments on Base chain. The x402 protocol handles payment automatically + using AgentKit's wallet provider for signing - no private keys exposed. + + Features: + - Access GPT-4o, Claude, Gemini, DeepSeek via single integration + - Pay-per-request in USDC (no monthly subscriptions) + - Uses AgentKit's wallet provider for secure signing + - Built on Coinbase's x402 protocol + """ + + def __init__(self, api_url: str | None = None): + """Initialize the BlockRun action provider. + + Args: + api_url: Optional custom API URL. Defaults to https://blockrun.ai/api/v1 + + """ + super().__init__("blockrun", []) + self._api_url = api_url or BLOCKRUN_API_URL + + def _get_session(self, wallet_provider: EvmWalletProvider): + """Create an x402-enabled requests session using the wallet provider. + + Args: + wallet_provider: The wallet provider for x402 payment signing. + + Returns: + x402 requests session configured with the wallet signer. + + """ + # Convert wallet provider to signer - no private key needed! + signer = wallet_provider.to_signer() + return x402_requests(signer) + + def _get_usdc_balance( + self, wallet_provider: EvmWalletProvider + ) -> tuple[float, str, str]: + """Get the USDC balance for the wallet. + + Args: + wallet_provider: The wallet provider to check balance for. + + Returns: + Tuple of (balance_float, formatted_balance, usdc_address). + + Raises: + ValueError: If the network is not supported or balance check fails. + + """ + from web3 import Web3 + + network = wallet_provider.get_network() + network_id = network.network_id + + if network_id not in USDC_ADDRESSES: + raise ValueError(f"USDC not configured for network: {network_id}") + + usdc_address = USDC_ADDRESSES[network_id] + wallet_address = wallet_provider.get_address() + + w3 = Web3() + checksum_usdc = w3.to_checksum_address(usdc_address) + checksum_wallet = w3.to_checksum_address(wallet_address) + + # Get decimals (USDC has 6 decimals) + decimals = wallet_provider.read_contract( + contract_address=checksum_usdc, + abi=ERC20_BALANCE_ABI, + function_name="decimals", + args=[], + ) + + # Get balance + balance_raw = wallet_provider.read_contract( + contract_address=checksum_usdc, + abi=ERC20_BALANCE_ABI, + function_name="balanceOf", + args=[checksum_wallet], + ) + + balance_float = balance_raw / (10**decimals) + formatted_balance = f"{balance_float:.6f} USDC" + + return balance_float, formatted_balance, usdc_address + + @create_action( + name="get_usdc_balance", + description=""" +Get the USDC balance for the wallet on Base chain. + +This action checks your wallet's USDC balance to ensure you have sufficient funds +for BlockRun API requests. BlockRun uses USDC micropayments on Base for pay-per-request +LLM access. + +Returns: +- Current USDC balance +- Wallet address +- Network information +- Funding instructions if balance is low""", + schema=GetUsdcBalanceSchema, + ) + def get_usdc_balance( + self, wallet_provider: EvmWalletProvider, args: dict[str, Any] + ) -> str: + """Get the USDC balance for the wallet. + + Args: + wallet_provider: The wallet provider to check balance for. + args: Empty dict (no parameters required). + + Returns: + str: JSON string containing balance information or error details. + + """ + try: + balance, formatted_balance, usdc_address = self._get_usdc_balance( + wallet_provider + ) + wallet_address = wallet_provider.get_address() + network = wallet_provider.get_network() + + result = { + "success": True, + "balance": balance, + "formatted_balance": formatted_balance, + "wallet_address": wallet_address, + "usdc_contract": usdc_address, + "network": network.network_id, + } + + # Add funding suggestion if balance is low + if balance < 0.10: + result["warning"] = "Low USDC balance" + result["suggestion"] = ( + "Your USDC balance is low. To fund your wallet:\n" + "1. Transfer USDC to your wallet on Base\n" + "2. Bridge USDC from another chain using a bridge like https://bridge.base.org\n" + "3. Buy USDC on Coinbase and withdraw to your wallet on Base" + ) + + return json.dumps(result, indent=2) + + except Exception as e: + return json.dumps( + { + "error": True, + "message": f"Failed to get USDC balance: {e!s}", + "suggestion": "Ensure you are connected to Base mainnet or Base Sepolia.", + }, + indent=2, + ) + + @create_action( + name="chat_completion", + description=""" +Send a chat completion request to an LLM via BlockRun using x402 micropayments. + +BlockRun provides access to multiple LLM providers with pay-per-request USDC payments +on Base chain. Payments are signed securely using your AgentKit wallet provider. + +Available models: +- openai/gpt-4o: Most capable GPT-4 model with vision capabilities +- openai/gpt-4o-mini: Fast and cost-effective GPT-4 model (default) +- anthropic/claude-sonnet-4: Anthropic's balanced model for most tasks +- google/gemini-2.0-flash: Google's fast multimodal model +- deepseek/deepseek-chat: DeepSeek's general-purpose chat model + +EXAMPLES: +- Simple question: chat_completion(prompt="What is the capital of France?") +- With system prompt: chat_completion(prompt="Write a poem", system_prompt="You are a creative poet") +- Using Claude: chat_completion(model="anthropic/claude-sonnet-4", prompt="Explain quantum computing") + +The payment is processed automatically via x402 - a small USDC fee is deducted per request.""", + schema=ChatCompletionSchema, + ) + def chat_completion( + self, wallet_provider: EvmWalletProvider, args: dict[str, Any] + ) -> str: + """Send a chat completion request via BlockRun. + + Args: + wallet_provider: The wallet provider for x402 payment signing. + args: Request parameters including model, prompt, system_prompt, etc. + + Returns: + str: JSON string containing the model's response or error details. + + """ + try: + # Check USDC balance before making the request + try: + balance, formatted_balance, _ = self._get_usdc_balance(wallet_provider) + if balance < 0.01: # Minimum balance threshold + return json.dumps( + { + "error": True, + "message": f"Insufficient USDC balance: {formatted_balance}", + "suggestion": ( + "Your wallet needs USDC on Base to pay for BlockRun requests. " + "Please fund your wallet with USDC:\n" + "1. Transfer USDC to your wallet on Base\n" + "2. Bridge USDC from another chain\n" + "3. Buy USDC on Coinbase and withdraw to Base" + ), + "wallet_address": wallet_provider.get_address(), + }, + indent=2, + ) + except Exception: + # If balance check fails, continue anyway - the x402 request will fail + # with a more specific error if there's actually a payment issue + pass + + # Build messages + messages = [] + if args.get("system_prompt"): + messages.append({"role": "system", "content": args["system_prompt"]}) + messages.append({"role": "user", "content": args["prompt"]}) + + # Create x402 session with wallet signer + session = self._get_session(wallet_provider) + + # Make the request - x402 handles payment automatically + response = session.post( + f"{self._api_url}/chat/completions", + json={ + "model": args.get("model", "openai/gpt-4o-mini"), + "messages": messages, + "max_tokens": args.get("max_tokens", 1024), + "temperature": args.get("temperature", 0.7), + }, + ) + + response.raise_for_status() + data = response.json() + + # Extract response content + content = data["choices"][0]["message"]["content"] + + return json.dumps( + { + "success": True, + "model": args.get("model", "openai/gpt-4o-mini"), + "response": content, + "usage": data.get("usage"), + "payment": "Paid via x402 micropayment on Base", + }, + indent=2, + ) + + except Exception as e: + error_message = str(e) + suggestion = "Check your wallet has USDC on Base and the model name is valid." + + # Provide more specific error messages + if "402" in error_message or "payment" in error_message.lower(): + suggestion = ( + "Payment failed. Ensure your wallet has sufficient USDC on Base. " + "You can check your balance and fund your wallet at https://blockrun.ai" + ) + elif "connection" in error_message.lower(): + suggestion = "Network error. Check your internet connection and try again." + + return json.dumps( + { + "error": True, + "message": f"BlockRun chat completion failed: {error_message}", + "suggestion": suggestion, + }, + indent=2, + ) + + @create_action( + name="list_models", + description=""" +List all available LLM models accessible via BlockRun. + +Returns information about each model including the provider, name, and description. +All models are accessible via pay-per-request USDC micropayments on Base chain.""", + schema=ListModelsSchema, + ) + def list_models(self, wallet_provider: EvmWalletProvider, args: dict[str, Any]) -> str: + """List available LLM models. + + Args: + wallet_provider: The wallet provider (not used for this action). + args: Empty dict (no parameters required). + + Returns: + str: JSON string containing available models. + + """ + return json.dumps( + { + "success": True, + "models": AVAILABLE_MODELS, + "payment_info": { + "network": "Base (Mainnet or Sepolia)", + "currency": "USDC", + "method": "x402 micropayments", + }, + }, + indent=2, + ) + + def supports_network(self, network: Network) -> bool: + """Check if the network is supported by this action provider. + + Args: + network: The network to check support for. + + Returns: + bool: Whether the network is supported. + + """ + return network.protocol_family == "evm" and network.network_id in SUPPORTED_NETWORKS + + +def blockrun_action_provider(api_url: str | None = None) -> BlockrunActionProvider: + """Create a new BlockRun action provider. + + BlockRun provides access to multiple LLM providers (OpenAI, Anthropic, Google, + DeepSeek) with pay-per-request USDC micropayments on Base chain using x402. + + Args: + api_url: Optional custom API URL. Defaults to https://blockrun.ai/api/v1 + + Returns: + BlockrunActionProvider: A new BlockRun action provider instance. + + Example: + ```python + from coinbase_agentkit import AgentKit, AgentKitConfig, blockrun_action_provider + + agentkit = AgentKit(AgentKitConfig( + wallet_provider=wallet_provider, + action_providers=[blockrun_action_provider()], + )) + ``` + + Learn more: https://blockrun.ai/docs + + """ + return BlockrunActionProvider(api_url=api_url) diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/schemas.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/schemas.py new file mode 100644 index 000000000..05687ff1f --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/blockrun/schemas.py @@ -0,0 +1,45 @@ +"""Schemas for BlockRun action provider.""" + +from pydantic import BaseModel, Field + + +class ChatCompletionSchema(BaseModel): + """Schema for chat completion request.""" + + model: str = Field( + default="openai/gpt-4o-mini", + description=( + "The model to use for chat completion. " + "Available models: openai/gpt-4o, openai/gpt-4o-mini, " + "anthropic/claude-sonnet-4, google/gemini-2.0-flash, " + "deepseek/deepseek-chat" + ), + ) + prompt: str = Field( + ..., + description="The user message or prompt to send to the model.", + ) + system_prompt: str | None = Field( + default=None, + description="Optional system prompt to set context for the conversation.", + ) + max_tokens: int = Field( + default=1024, + description="Maximum number of tokens to generate in the response.", + ) + temperature: float = Field( + default=0.7, + description="Sampling temperature between 0 and 2. Higher values make output more random.", + ) + + +class ListModelsSchema(BaseModel): + """Schema for listing available models.""" + + pass + + +class GetUsdcBalanceSchema(BaseModel): + """Schema for getting USDC balance.""" + + pass diff --git a/python/coinbase-agentkit/pyproject.toml b/python/coinbase-agentkit/pyproject.toml index 0b6ec945c..9ba0918d5 100644 --- a/python/coinbase-agentkit/pyproject.toml +++ b/python/coinbase-agentkit/pyproject.toml @@ -31,6 +31,9 @@ dependencies = [ "solders>=0.26.0" ] +[project.optional-dependencies] +# BlockRun uses built-in x402 support - no extra dependencies needed + [tool.hatch.metadata] allow-direct-references = true diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/__init__.py b/python/coinbase-agentkit/tests/action_providers/blockrun/__init__.py new file mode 100644 index 000000000..50b54c0ce --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/__init__.py @@ -0,0 +1 @@ +"""Tests for BlockRun action provider.""" diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/conftest.py b/python/coinbase-agentkit/tests/action_providers/blockrun/conftest.py new file mode 100644 index 000000000..8949bb1d7 --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/conftest.py @@ -0,0 +1,83 @@ +"""Test fixtures for BlockRun action provider.""" + +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture +def mock_wallet_provider(): + """Create a mock wallet provider for testing. + + The wallet provider has a to_signer() method that returns a signer + object compatible with the x402 library. + """ + mock_provider = MagicMock() + mock_signer = MagicMock() + mock_signer.address = "0x1234567890123456789012345678901234567890" + mock_provider.to_signer.return_value = mock_signer + mock_provider.get_address.return_value = "0x1234567890123456789012345678901234567890" + return mock_provider + + +@pytest.fixture +def mock_x402_session(): + """Create a mock x402 session for testing.""" + mock_session = MagicMock() + + # Setup mock response + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "id": "chatcmpl-123", + "object": "chat.completion", + "created": 1677652288, + "model": "openai/gpt-4o-mini", + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": "This is a test response from the LLM.", + }, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 20, + "total_tokens": 30, + }, + } + mock_response.raise_for_status = MagicMock() + + mock_session.post.return_value = mock_response + mock_session.get.return_value = mock_response + + return mock_session + + +@pytest.fixture +def provider(mock_x402_session): + """Create a BlockrunActionProvider with mocked x402 session. + + Args: + mock_x402_session: Mock x402 session for HTTP requests. + + Returns: + BlockrunActionProvider: Provider configured for testing. + + """ + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + provider = BlockrunActionProvider() + # Store the mock for use in tests + provider._mock_x402_requests = mock_x402_requests + provider._mock_session = mock_x402_session + yield provider diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/__init__.py b/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/__init__.py new file mode 100644 index 000000000..795724c0d --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/__init__.py @@ -0,0 +1 @@ +"""End-to-end tests for BlockRun action provider.""" diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/conftest.py b/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/conftest.py new file mode 100644 index 000000000..944ee2b75 --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/conftest.py @@ -0,0 +1,58 @@ +"""E2E test fixtures for BlockRun action provider.""" + +import os + +import pytest + + +@pytest.fixture +def cdp_wallet_provider(): + """Create a real CDP wallet provider for e2e testing. + + Requires environment variables: + - CDP_API_KEY_ID + - CDP_API_KEY_SECRET + - CDP_WALLET_SECRET + + The wallet should have USDC on Base mainnet for testing. + """ + api_key_id = os.environ.get("CDP_API_KEY_ID") + api_key_secret = os.environ.get("CDP_API_KEY_SECRET") + wallet_secret = os.environ.get("CDP_WALLET_SECRET") + wallet_address = os.environ.get("CDP_WALLET_ADDRESS") # Optional + + if not all([api_key_id, api_key_secret, wallet_secret]): + pytest.skip( + "CDP credentials not set. Set CDP_API_KEY_ID, CDP_API_KEY_SECRET, " + "and CDP_WALLET_SECRET environment variables." + ) + + from coinbase_agentkit.wallet_providers import ( + CdpEvmWalletProvider, + CdpEvmWalletProviderConfig, + ) + + config = CdpEvmWalletProviderConfig( + api_key_id=api_key_id, + api_key_secret=api_key_secret, + wallet_secret=wallet_secret, + network_id="base-mainnet", + address=wallet_address, + ) + + return CdpEvmWalletProvider(config) + + +@pytest.fixture +def e2e_provider(): + """Create a BlockrunActionProvider for e2e testing. + + Returns: + BlockrunActionProvider: Provider configured for real API calls. + + """ + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + return BlockrunActionProvider() diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/test_chat_completion_e2e.py b/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/test_chat_completion_e2e.py new file mode 100644 index 000000000..03e26676b --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/e2e/test_chat_completion_e2e.py @@ -0,0 +1,73 @@ +"""End-to-end tests for BlockRun chat_completion action. + +These tests make real API calls and require: +- CDP_API_KEY_ID, CDP_API_KEY_SECRET, CDP_WALLET_SECRET environment variables +- Wallet with USDC balance on Base mainnet + +Run with: pytest -m e2e tests/action_providers/blockrun/e2e/ +""" + +import json + +import pytest + + +@pytest.mark.e2e +def test_chat_completion_real_api(e2e_provider, cdp_wallet_provider): + """Test real chat completion API call.""" + args = { + "prompt": "What is 2 + 2? Reply with just the number.", + "model": "openai/gpt-4o-mini", + "max_tokens": 10, + "temperature": 0.0, + } + + result = e2e_provider.chat_completion(cdp_wallet_provider, args) + result_data = json.loads(result) + + print(f"E2E Result: {json.dumps(result_data, indent=2)}") + + assert result_data["success"] is True + assert "response" in result_data + assert "4" in result_data["response"] + assert result_data["payment"] == "Paid via x402 micropayment on Base" + + +@pytest.mark.e2e +def test_chat_completion_claude(e2e_provider, cdp_wallet_provider): + """Test real chat completion with Claude.""" + args = { + "prompt": "Say 'Hello BlockRun' and nothing else.", + "model": "anthropic/claude-sonnet-4", + "max_tokens": 20, + "temperature": 0.0, + } + + result = e2e_provider.chat_completion(cdp_wallet_provider, args) + result_data = json.loads(result) + + print(f"Claude E2E Result: {json.dumps(result_data, indent=2)}") + + assert result_data["success"] is True + assert "response" in result_data + assert "BlockRun" in result_data["response"] or "Hello" in result_data["response"] + + +@pytest.mark.e2e +def test_chat_completion_with_system_prompt(e2e_provider, cdp_wallet_provider): + """Test real chat completion with system prompt.""" + args = { + "prompt": "What are you?", + "model": "openai/gpt-4o-mini", + "system_prompt": "You are a helpful pirate. Always talk like a pirate.", + "max_tokens": 50, + "temperature": 0.7, + } + + result = e2e_provider.chat_completion(cdp_wallet_provider, args) + result_data = json.loads(result) + + print(f"System Prompt E2E Result: {json.dumps(result_data, indent=2)}") + + assert result_data["success"] is True + assert "response" in result_data diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/test_blockrun_action_provider.py b/python/coinbase-agentkit/tests/action_providers/blockrun/test_blockrun_action_provider.py new file mode 100644 index 000000000..33c7eae8e --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/test_blockrun_action_provider.py @@ -0,0 +1,123 @@ +"""Tests for the BlockRun action provider initialization.""" + +from unittest.mock import MagicMock, patch + +from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + AVAILABLE_MODELS, + BLOCKRUN_API_URL, + SUPPORTED_NETWORKS, + BlockrunActionProvider, + blockrun_action_provider, +) +from coinbase_agentkit.network import Network + + +def test_init_default(): + """Test initialization with defaults.""" + provider = BlockrunActionProvider() + assert provider is not None + assert provider._api_url == BLOCKRUN_API_URL + + +def test_init_with_custom_api_url(): + """Test initialization with custom API URL.""" + custom_url = "https://custom.blockrun.ai/api/v1" + provider = BlockrunActionProvider(api_url=custom_url) + assert provider._api_url == custom_url + + +def test_get_session_uses_wallet_provider(mock_wallet_provider): + """Test that _get_session uses wallet provider's signer.""" + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_session = MagicMock() + mock_x402_requests.return_value = mock_session + + provider = BlockrunActionProvider() + session = provider._get_session(mock_wallet_provider) + + # Verify wallet provider's to_signer was called + mock_wallet_provider.to_signer.assert_called_once() + + # Verify x402_requests was called with the signer + mock_x402_requests.assert_called_once() + assert session == mock_session + + +def test_supports_network_base_mainnet(): + """Test supports_network for Base Mainnet.""" + provider = BlockrunActionProvider() + network = Network( + name="base-mainnet", + protocol_family="evm", + chain_id="8453", + network_id="base-mainnet", + ) + assert provider.supports_network(network) is True + + +def test_supports_network_base_sepolia(): + """Test supports_network for Base Sepolia.""" + provider = BlockrunActionProvider() + network = Network( + name="base-sepolia", + protocol_family="evm", + chain_id="84532", + network_id="base-sepolia", + ) + assert provider.supports_network(network) is True + + +def test_supports_network_unsupported(): + """Test supports_network for unsupported network.""" + provider = BlockrunActionProvider() + network = Network( + name="ethereum-mainnet", + protocol_family="evm", + chain_id="1", + network_id="ethereum-mainnet", + ) + assert provider.supports_network(network) is False + + +def test_supports_network_non_evm(): + """Test supports_network for non-EVM network.""" + provider = BlockrunActionProvider() + network = Network( + name="solana-mainnet", + protocol_family="solana", + chain_id="", + network_id="solana-mainnet", + ) + assert provider.supports_network(network) is False + + +def test_factory_function(): + """Test the factory function.""" + provider = blockrun_action_provider() + assert isinstance(provider, BlockrunActionProvider) + assert provider._api_url == BLOCKRUN_API_URL + + +def test_factory_function_with_custom_url(): + """Test the factory function with custom API URL.""" + custom_url = "https://custom.blockrun.ai/api/v1" + provider = blockrun_action_provider(api_url=custom_url) + assert isinstance(provider, BlockrunActionProvider) + assert provider._api_url == custom_url + + +def test_available_models(): + """Test that available models are defined correctly.""" + assert "openai/gpt-4o" in AVAILABLE_MODELS + assert "openai/gpt-4o-mini" in AVAILABLE_MODELS + assert "anthropic/claude-sonnet-4" in AVAILABLE_MODELS + assert "google/gemini-2.0-flash" in AVAILABLE_MODELS + assert "deepseek/deepseek-chat" in AVAILABLE_MODELS + + +def test_supported_networks(): + """Test that supported networks are defined correctly.""" + assert "base-mainnet" in SUPPORTED_NETWORKS + assert "base-sepolia" in SUPPORTED_NETWORKS diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/test_chat_completion.py b/python/coinbase-agentkit/tests/action_providers/blockrun/test_chat_completion.py new file mode 100644 index 000000000..4b8bc6719 --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/test_chat_completion.py @@ -0,0 +1,222 @@ +"""Tests for BlockRun chat_completion action.""" + +import json +from unittest.mock import patch + + +def test_chat_completion_basic(mock_wallet_provider, mock_x402_session): + """Test basic chat completion request.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = {"prompt": "Hello, how are you?"} + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["success"] is True + assert "response" in result_data + assert result_data["model"] == "openai/gpt-4o-mini" # default model + assert "payment" in result_data + + # Verify the session.post was called with correct URL + mock_x402_session.post.assert_called_once() + call_args = mock_x402_session.post.call_args + assert "chat/completions" in call_args[0][0] + + +def test_chat_completion_with_model(mock_wallet_provider, mock_x402_session): + """Test chat completion with specific model.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = { + "prompt": "Explain quantum computing", + "model": "anthropic/claude-sonnet-4", + } + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["success"] is True + assert result_data["model"] == "anthropic/claude-sonnet-4" + + # Verify the model was passed in the request + call_args = mock_x402_session.post.call_args + assert call_args[1]["json"]["model"] == "anthropic/claude-sonnet-4" + + +def test_chat_completion_with_system_prompt(mock_wallet_provider, mock_x402_session): + """Test chat completion with system prompt.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = { + "prompt": "Write a haiku", + "system_prompt": "You are a creative poet.", + } + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["success"] is True + + # Verify system prompt was included in messages + call_args = mock_x402_session.post.call_args + messages = call_args[1]["json"]["messages"] + assert len(messages) == 2 + assert messages[0]["role"] == "system" + assert messages[0]["content"] == "You are a creative poet." + assert messages[1]["role"] == "user" + + +def test_chat_completion_with_parameters(mock_wallet_provider, mock_x402_session): + """Test chat completion with all parameters.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = { + "prompt": "Tell me a joke", + "model": "openai/gpt-4o", + "system_prompt": "You are a comedian.", + "max_tokens": 500, + "temperature": 0.9, + } + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["success"] is True + assert result_data["model"] == "openai/gpt-4o" + + # Verify all parameters were passed + call_args = mock_x402_session.post.call_args + request_json = call_args[1]["json"] + assert request_json["max_tokens"] == 500 + assert request_json["temperature"] == 0.9 + + +def test_chat_completion_includes_usage(mock_wallet_provider, mock_x402_session): + """Test that chat completion includes token usage.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = {"prompt": "Hello"} + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["success"] is True + assert "usage" in result_data + assert result_data["usage"]["prompt_tokens"] == 10 + assert result_data["usage"]["completion_tokens"] == 20 + assert result_data["usage"]["total_tokens"] == 30 + + +def test_chat_completion_error_handling(mock_wallet_provider, mock_x402_session): + """Test chat completion error handling.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + # Make the session raise an exception + mock_x402_session.post.side_effect = Exception("API error") + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = {"prompt": "Hello"} + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["error"] is True + assert "API error" in result_data["message"] + assert "suggestion" in result_data + + +def test_chat_completion_payment_error(mock_wallet_provider, mock_x402_session): + """Test chat completion with payment error.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + # Make the session raise a payment error + mock_x402_session.post.side_effect = Exception("402 Payment Required") + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = {"prompt": "Hello"} + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["error"] is True + assert "USDC" in result_data["suggestion"] + + +def test_chat_completion_uses_wallet_provider_signer(mock_wallet_provider, mock_x402_session): + """Test that chat completion uses wallet provider's signer for x402.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = {"prompt": "Hello"} + + provider.chat_completion(mock_wallet_provider, args) + + # Verify wallet provider's to_signer was called + mock_wallet_provider.to_signer.assert_called_once() + + # Verify x402_requests was called with the signer + mock_x402_requests.assert_called_once() + signer_arg = mock_x402_requests.call_args[0][0] + assert signer_arg == mock_wallet_provider.to_signer.return_value diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/test_get_usdc_balance.py b/python/coinbase-agentkit/tests/action_providers/blockrun/test_get_usdc_balance.py new file mode 100644 index 000000000..2ed6648ed --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/test_get_usdc_balance.py @@ -0,0 +1,199 @@ +"""Tests for BlockRun get_usdc_balance action.""" + +import json +from unittest.mock import MagicMock, patch + +from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + USDC_ADDRESSES, + BlockrunActionProvider, +) +from coinbase_agentkit.network import Network + + +def test_get_usdc_balance_success(): + """Test successful USDC balance check.""" + mock_provider = MagicMock() + mock_provider.get_address.return_value = "0x1234567890123456789012345678901234567890" + mock_provider.get_network.return_value = Network( + name="base-mainnet", + protocol_family="evm", + chain_id="8453", + network_id="base-mainnet", + ) + + # Mock contract reads: decimals (6) and balance (1000000 = 1 USDC) + mock_provider.read_contract.side_effect = [6, 1000000] + + provider = BlockrunActionProvider() + result = provider.get_usdc_balance(mock_provider, {}) + result_data = json.loads(result) + + assert result_data["success"] is True + assert result_data["balance"] == 1.0 + assert result_data["formatted_balance"] == "1.000000 USDC" + assert result_data["network"] == "base-mainnet" + assert result_data["usdc_contract"] == USDC_ADDRESSES["base-mainnet"] + + +def test_get_usdc_balance_low_balance_warning(): + """Test low balance warning.""" + mock_provider = MagicMock() + mock_provider.get_address.return_value = "0x1234567890123456789012345678901234567890" + mock_provider.get_network.return_value = Network( + name="base-mainnet", + protocol_family="evm", + chain_id="8453", + network_id="base-mainnet", + ) + + # Mock contract reads: decimals (6) and balance (50000 = 0.05 USDC) + mock_provider.read_contract.side_effect = [6, 50000] + + provider = BlockrunActionProvider() + result = provider.get_usdc_balance(mock_provider, {}) + result_data = json.loads(result) + + assert result_data["success"] is True + assert result_data["balance"] == 0.05 + assert result_data["warning"] == "Low USDC balance" + assert "suggestion" in result_data + + +def test_get_usdc_balance_base_sepolia(): + """Test USDC balance check on Base Sepolia.""" + mock_provider = MagicMock() + mock_provider.get_address.return_value = "0x1234567890123456789012345678901234567890" + mock_provider.get_network.return_value = Network( + name="base-sepolia", + protocol_family="evm", + chain_id="84532", + network_id="base-sepolia", + ) + + # Mock contract reads: decimals (6) and balance (5000000 = 5 USDC) + mock_provider.read_contract.side_effect = [6, 5000000] + + provider = BlockrunActionProvider() + result = provider.get_usdc_balance(mock_provider, {}) + result_data = json.loads(result) + + assert result_data["success"] is True + assert result_data["balance"] == 5.0 + assert result_data["network"] == "base-sepolia" + assert result_data["usdc_contract"] == USDC_ADDRESSES["base-sepolia"] + + +def test_get_usdc_balance_unsupported_network(): + """Test USDC balance check on unsupported network.""" + mock_provider = MagicMock() + mock_provider.get_address.return_value = "0x1234567890123456789012345678901234567890" + mock_provider.get_network.return_value = Network( + name="ethereum-mainnet", + protocol_family="evm", + chain_id="1", + network_id="ethereum-mainnet", + ) + + provider = BlockrunActionProvider() + result = provider.get_usdc_balance(mock_provider, {}) + result_data = json.loads(result) + + assert result_data["error"] is True + assert "not configured" in result_data["message"] + + +def test_get_usdc_balance_contract_error(): + """Test USDC balance check with contract error.""" + mock_provider = MagicMock() + mock_provider.get_address.return_value = "0x1234567890123456789012345678901234567890" + mock_provider.get_network.return_value = Network( + name="base-mainnet", + protocol_family="evm", + chain_id="8453", + network_id="base-mainnet", + ) + mock_provider.read_contract.side_effect = Exception("Contract call failed") + + provider = BlockrunActionProvider() + result = provider.get_usdc_balance(mock_provider, {}) + result_data = json.loads(result) + + assert result_data["error"] is True + assert "Failed to get USDC balance" in result_data["message"] + + +def test_chat_completion_insufficient_balance(mock_wallet_provider, mock_x402_session): + """Test chat completion with insufficient USDC balance.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + # Setup mock wallet provider with network and low balance + mock_wallet_provider.get_network.return_value = Network( + name="base-mainnet", + protocol_family="evm", + chain_id="8453", + network_id="base-mainnet", + ) + # Mock contract reads: decimals (6) and balance (5000 = 0.005 USDC - too low) + mock_wallet_provider.read_contract.side_effect = [6, 5000] + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = {"prompt": "Hello"} + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["error"] is True + assert "Insufficient USDC balance" in result_data["message"] + # The x402 session should NOT have been called + mock_x402_session.post.assert_not_called() + + +def test_chat_completion_proceeds_with_sufficient_balance( + mock_wallet_provider, mock_x402_session +): + """Test chat completion proceeds when USDC balance is sufficient.""" + from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + BlockrunActionProvider, + ) + + # Setup mock wallet provider with network and sufficient balance + mock_wallet_provider.get_network.return_value = Network( + name="base-mainnet", + protocol_family="evm", + chain_id="8453", + network_id="base-mainnet", + ) + # Mock contract reads: decimals (6) and balance (1000000 = 1 USDC) + mock_wallet_provider.read_contract.side_effect = [6, 1000000] + + with patch( + "coinbase_agentkit.action_providers.blockrun.blockrun_action_provider.x402_requests" + ) as mock_x402_requests: + mock_x402_requests.return_value = mock_x402_session + + provider = BlockrunActionProvider() + args = {"prompt": "Hello"} + + result = provider.chat_completion(mock_wallet_provider, args) + result_data = json.loads(result) + + assert result_data["success"] is True + # The x402 session SHOULD have been called + mock_x402_session.post.assert_called_once() + + +def test_usdc_addresses_defined(): + """Test that USDC addresses are defined for supported networks.""" + assert "base-mainnet" in USDC_ADDRESSES + assert "base-sepolia" in USDC_ADDRESSES + # Verify address format (0x + 40 hex chars) + for _network, address in USDC_ADDRESSES.items(): + assert address.startswith("0x") + assert len(address) == 42 diff --git a/python/coinbase-agentkit/tests/action_providers/blockrun/test_list_models.py b/python/coinbase-agentkit/tests/action_providers/blockrun/test_list_models.py new file mode 100644 index 000000000..aa62b50f0 --- /dev/null +++ b/python/coinbase-agentkit/tests/action_providers/blockrun/test_list_models.py @@ -0,0 +1,76 @@ +"""Tests for BlockRun list_models action.""" + +import json + +from coinbase_agentkit.action_providers.blockrun.blockrun_action_provider import ( + AVAILABLE_MODELS, + BlockrunActionProvider, +) + + +def test_list_models(mock_wallet_provider): + """Test list_models action.""" + provider = BlockrunActionProvider() + + result = provider.list_models(mock_wallet_provider, {}) + result_data = json.loads(result) + + assert result_data["success"] is True + assert "models" in result_data + assert "payment_info" in result_data + + +def test_list_models_contains_all_models(mock_wallet_provider): + """Test that list_models contains all available models.""" + provider = BlockrunActionProvider() + + result = provider.list_models(mock_wallet_provider, {}) + result_data = json.loads(result) + + models = result_data["models"] + + # Check all expected models are present + assert "openai/gpt-4o" in models + assert "openai/gpt-4o-mini" in models + assert "anthropic/claude-sonnet-4" in models + assert "google/gemini-2.0-flash" in models + assert "deepseek/deepseek-chat" in models + + +def test_list_models_model_structure(mock_wallet_provider): + """Test that models have correct structure.""" + provider = BlockrunActionProvider() + + result = provider.list_models(mock_wallet_provider, {}) + result_data = json.loads(result) + + models = result_data["models"] + + for _model_id, model_info in models.items(): + assert "name" in model_info + assert "provider" in model_info + assert "description" in model_info + + +def test_list_models_payment_info(mock_wallet_provider): + """Test that payment_info is correct.""" + provider = BlockrunActionProvider() + + result = provider.list_models(mock_wallet_provider, {}) + result_data = json.loads(result) + + payment_info = result_data["payment_info"] + + assert payment_info["network"] == "Base (Mainnet or Sepolia)" + assert payment_info["currency"] == "USDC" + assert payment_info["method"] == "x402 micropayments" + + +def test_list_models_matches_constant(mock_wallet_provider): + """Test that list_models returns same models as AVAILABLE_MODELS constant.""" + provider = BlockrunActionProvider() + + result = provider.list_models(mock_wallet_provider, {}) + result_data = json.loads(result) + + assert result_data["models"] == AVAILABLE_MODELS