pythoncoinbase-api

How can I use COINBase RESTClient


I went to coinbase and registered an API key

https://docs.cdp.coinbase.com/coinbase-app/docs/getting-started

[!api_definedhttps://cloud.coinbase.com/access/api]1

I put the API Key and Secret in a .env file

these are not the real secret or api key

COINBASE_API_KEY=3398a3
COINBASE_API_SECRET=-----BEGIN EC PRIVATE KEY-----
EsOY8CRow==
-----END EC PRIVATE KEY-----

Requirements.txt

fastapi
uvicorn[standard]


# for environment variables
python-dotenv


# Coinbase Advanced Trade API client
coinbase-advanced-py

Here is the server code:

from fastapi import FastAPI
from app.endpoints import account

from dotenv import load_dotenv
load_dotenv()  # Loads .env file into environment variables

import logging
logging.basicConfig(
    level=logging.INFO,  # <-- enable info-level logs
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

app = FastAPI(
    title="crypto-trader",
    description="A service for interacting with Coinbase's advanced trade API.",
    version="1.0.0"
)

app.include_router(account.router)

here is the account endpoint

# app/endpoints/account.py
from fastapi import APIRouter
from app.services.coinbase_client import get_accounts

router = APIRouter()

@router.get("/account", tags=["Account"])
def account_info():
    return get_accounts()

here is the code that accesses the coin base account

# app/services/coinbase_client.py
import logging
from app.core.config import COINBASE_API_KEY, COINBASE_API_SECRET

logger = logging.getLogger(__name__)

try:
    from coinbase.rest import RESTClient
except ImportError as e:
    logger.error("Failed to import RESTClient from coinbase.rest")
    logger.exception(e)
    raise

try:
    client = RESTClient(api_key=COINBASE_API_KEY, api_secret=COINBASE_API_SECRET, verbose=True)
except Exception as e:
    logger.error("Failed to initialize RESTClient with provided API credentials")
    logger.exception(e)
    raise

def get_accounts():
    logger.info(f"==================================================COINBASE_API_KEY: {COINBASE_API_KEY}")
    try:
        return client.get_accounts()
    except Exception as e:
        logger.error("Error fetching profiles from Coinbase API")
        logger.exception(e)
        # Option 1: Raise so caller can handle the error explicitly
        # raise

        # Option 2: Return an error dict (your original choice)
        return {"error": str(e)}

when I start the server I see this, so the server "runs"

[!server-runsenter image description here]2

however when I hit the endpoint I get this:

2025-06-12 00:05:57,379 - app.services.coinbase_client - INFO - ==================================================COINBASE_API_KEY: 3398a3
2025-06-12 00:05:57,381 - app.services.coinbase_client - ERROR - Error fetching profiles from Coinbase API
2025-06-12 00:05:57,381 - app.services.coinbase_client - ERROR - Unable to load PEM file. See https://cryptography.io/en/latest/faq/#why-can-t-i-import-my-pem-file for more details. MalformedFraming
Are you sure you generated your key at https://cloud.coinbase.com/access/api ?
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/coinbase/jwt_generator.py", line 16, in build_jwt
    private_key = serialization.load_pem_private_key(
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: Unable to load PEM file. See https://cryptography.io/en/latest/faq/#why-can-t-i-import-my-pem-file for more details. MalformedFraming

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/workspace/app/services/coinbase_client.py", line 29, in get_accounts
    return client.get_accounts()
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/coinbase/rest/accounts.py", line 37, in get_accounts
    return ListAccountsResponse(self.get(endpoint, params=params, **kwargs))
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/coinbase/rest/rest_base.py", line 101, in get
    return self.prepare_and_send_request(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/coinbase/rest/rest_base.py", line 199, in prepare_and_send_request
    headers = self.set_headers(http_method, url_path)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/coinbase/rest/rest_base.py", line 255, in set_headers
    "Authorization": f"Bearer {jwt_generator.build_rest_jwt(uri, self.api_key, self.api_secret)}",
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/coinbase/jwt_generator.py", line 63, in build_rest_jwt
    return build_jwt(key_var, secret_var, uri=uri)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/coinbase/jwt_generator.py", line 21, in build_jwt
    raise Exception(
Exception: Unable to load PEM file. See https://cryptography.io/en/latest/faq/#why-can-t-i-import-my-pem-file for more details. MalformedFraming
Are you sure you generated your key at https://cloud.coinbase.com/access/api ?

Solution

  • here is the answer:

    # app/services/jwt_helper.py
    
    import os
    import base64
    import logging
    
    from coinbase import jwt_generator
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives import serialization
    
    logger = logging.getLogger(__name__)
    
    COINBASE_API_KEY = os.getenv("COINBASE_API_KEY")
    COINBASE_API_SECRET_B64 = os.getenv("COINBASE_API_SECRET")
    
    def to_pem(secret_b64: str) -> str:
        try:
            raw_bytes = base64.b64decode(secret_b64)
    
            # If too long, slice the first 32 bytes assuming they are the scalar
            if len(raw_bytes) == 64:
                raw_bytes = raw_bytes[:32]
    
            if len(raw_bytes) != 32:
                raise ValueError(f"Expected 32 bytes for SECP256K1 private key, got {len(raw_bytes)} bytes")
    
            priv_int = int.from_bytes(raw_bytes, "big")
    
            priv_key = ec.derive_private_key(priv_int, ec.SECP256K1())
            pem = priv_key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()
            )
            return pem.decode("utf-8")
    
        except Exception as e:
            logger.error(f"❌ Failed to convert secret to PEM: {e}")
            raise
    
    
    
    pem_private_key = to_pem(COINBASE_API_SECRET_B64)
    
    def generate_jwt_token(
        request_method: str = "GET",
        request_path: str = "/v2/accounts"
    ) -> str:
        logger.info(f"=============COINBASE_API_KEY: {COINBASE_API_KEY}")
        logger.info(f"=============PEM_wrapped_secret: {pem_private_key}")
    
        jwt_uri = jwt_generator.format_jwt_uri(request_method, request_path)
        jwt_token = jwt_generator.build_rest_jwt(jwt_uri, COINBASE_API_KEY, pem_private_key)
        return jwt_token