feat: Pyth Network Price Feed Actions

cdp-agentkit-core/python/CHANGELOG.md

@@ -6,6 +6,8 @@
 
 - Added `get_balance_nft` action.
 - Added `transfer_nft` action.
+- Added `pyth_fetch_price_feed_id` action to fetch the price feed ID for a given token symbol from Pyth.
+- Added `pyth_fetch_price` action to fetch the price of a given price feed from Pyth.
 
 ## [0.0.8] - 2025-01-13
 

cdp-agentkit-core/python/cdp_agentkit_core/actions/__init__.py

@@ -5,6 +5,8 @@
 from cdp_agentkit_core.actions.get_balance_nft import GetBalanceNftAction
 from cdp_agentkit_core.actions.get_wallet_details import GetWalletDetailsAction
 from cdp_agentkit_core.actions.mint_nft import MintNftAction
+from cdp_agentkit_core.actions.pyth.fetch_price import PythFetchPriceAction
+from cdp_agentkit_core.actions.pyth.fetch_price_feed_id import PythFetchPriceFeedIDAction
 from cdp_agentkit_core.actions.register_basename import RegisterBasenameAction
 from cdp_agentkit_core.actions.request_faucet_funds import RequestFaucetFundsAction
 from cdp_agentkit_core.actions.trade import TradeAction
@@ -46,4 +48,6 @@ def get_all_cdp_actions() -> list[type[CdpAction]]:
     "WowCreateTokenAction",
     "WowSellTokenAction",
     "WrapEthAction",
+    "PythFetchPriceFeedIDAction",
+    "PythFetchPriceAction",
 ]

cdp-agentkit-core/python/cdp_agentkit_core/actions/pyth/fetch_price.py

@@ -0,0 +1,60 @@
+from collections.abc import Callable
+
+import requests
+from pydantic import BaseModel, Field
+
+from cdp_agentkit_core.actions import CdpAction
+
+PYTH_FETCH_PRICE_PROMPT = """
+Fetch the price of a given price feed from Pyth. First fetch the price feed ID forusing the pyth_fetch_price_feed_id action.
+
+Inputs:
+- Pyth price feed ID
+
+Important notes:
+- Do not assume that a random ID is a Pyth price feed ID. If you are confused, ask a clarifying question.
+- This action only fetches price inputs from Pyth price feeds. No other source.
+- If you are asked to fetch the price from Pyth for a ticker symbol such as BTC, you must first use the pyth_fetch_price_feed_id
+action to retrieve the price feed ID before invoking the pyth_Fetch_price action
+"""
+
+
+class PythFetchPriceInput(BaseModel):
+    """Input schema for fetching Pyth price."""
+
+    price_feed_id: str = Field(..., description="The price feed ID to fetch the price for.")
+
+
+def pyth_fetch_price(price_feed_id: str) -> str:
+    """Fetch the price of a given price feed from Pyth."""
+    url = f"https://hermes.pyth.network/v2/updates/price/latest?ids[]={price_feed_id}"
+    response = requests.get(url)
+    response.raise_for_status()
+    data = response.json()
+    parsed_data = data["parsed"]
+
+    if not parsed_data:
+        raise ValueError(f"No price data found for {price_feed_id}")
+
+    price_info = parsed_data[0]["price"]
+    price = int(price_info["price"])
+    exponent = price_info["expo"]
+
+    if exponent < 0:
+        adjusted_price = price * 100
+        divisor = 10**-exponent
+        scaled_price = adjusted_price // divisor
+        price_str = f"{scaled_price // 100}.{scaled_price % 100:02}"
+        return price_str if not price_str.startswith(".") else f"0{price_str}"
+
+    scaled_price = price // (10**exponent)
+    return str(scaled_price)
+
+
+class PythFetchPriceAction(CdpAction):
+    """Fetch Pyth Price action."""
+
+    name: str = "pyth_fetch_price"
+    description: str = PYTH_FETCH_PRICE_PROMPT
+    args_schema: type[BaseModel] | None = PythFetchPriceInput
+    func: Callable[..., str] = pyth_fetch_price

cdp-agentkit-core/python/cdp_agentkit_core/actions/pyth/fetch_price_feed_id.py

@@ -0,0 +1,44 @@
+from collections.abc import Callable
+
+import requests
+from pydantic import BaseModel, Field
+
+from cdp_agentkit_core.actions import CdpAction
+
+PYTH_FETCH_PRICE_FEED_ID_PROMPT = """
+Fetch the price feed ID for a given token symbol (e.g. BTC, ETH, etc.) from Pyth.
+"""
+
+
+class PythFetchPriceFeedIDInput(BaseModel):
+    """Input schema for fetching Pyth price feed ID."""
+
+    token_symbol: str = Field(..., description="The token symbol to fetch the price feed ID for.")
+
+
+def pyth_fetch_price_feed_id(token_symbol: str) -> str:
+    """Fetch the price feed ID for a given token symbol from Pyth."""
+    url = f"https://hermes.pyth.network/v2/price_feeds?query={token_symbol}&asset_type=crypto"
+    response = requests.get(url)
+    response.raise_for_status()
+    data = response.json()
+
+    if not data:
+        raise ValueError(f"No price feed found for {token_symbol}")
+
+    filtered_data = [
+        item for item in data if item["attributes"]["base"].lower() == token_symbol.lower()
+    ]
+    if not filtered_data:
+        raise ValueError(f"No price feed found for {token_symbol}")
+
+    return filtered_data[0]["id"]
+
+
+class PythFetchPriceFeedIDAction(CdpAction):
+    """Pyth Fetch Price Feed ID action."""
+
+    name: str = "pyth_fetch_price_feed_id"
+    description: str = PYTH_FETCH_PRICE_FEED_ID_PROMPT
+    args_schema: type[BaseModel] | None = PythFetchPriceFeedIDInput
+    func: Callable[..., str] = pyth_fetch_price_feed_id

cdp-agentkit-core/python/tests/actions/pyth/test_fetch_price.py

@@ -0,0 +1,61 @@
+from unittest.mock import patch
+
+import pytest
+import requests
+
+from cdp_agentkit_core.actions.pyth.fetch_price import (
+    PythFetchPriceInput,
+    pyth_fetch_price,
+)
+
+MOCK_PRICE_FEED_ID = "valid-price-feed-id"
+
+
+def test_pyth_fetch_price_input_model_valid():
+    """Test that PythFetchPriceInput accepts valid parameters."""
+    input_model = PythFetchPriceInput(
+        price_feed_id=MOCK_PRICE_FEED_ID,
+    )
+
+    assert input_model.price_feed_id == MOCK_PRICE_FEED_ID
+
+
+def test_pyth_fetch_price_input_model_missing_params():
+    """Test that PythFetchPriceInput raises error when params are missing."""
+    with pytest.raises(ValueError):
+        PythFetchPriceInput()
+
+
+def test_pyth_fetch_price_success():
+    """Test successful pyth fetch price with valid parameters."""
+    mock_response = {
+        "parsed": [
+            {
+                "price": {
+                    "price": "4212345",
+                    "expo": -2,
+                    "conf": "1234",
+                },
+                "id": "test_feed_id",
+            }
+        ]
+    }
+
+    with patch("requests.get") as mock_get:
+        mock_get.return_value.json.return_value = mock_response
+        mock_get.return_value.raise_for_status.return_value = None
+
+        result = pyth_fetch_price(MOCK_PRICE_FEED_ID)
+
+        assert result == "42123.45"
+
+
+def test_pyth_fetch_price_http_error():
+    """Test pyth fetch price error with HTTP error."""
+    with patch("requests.get") as mock_get:
+        mock_get.return_value.raise_for_status.side_effect = requests.exceptions.HTTPError(
+            "404 Client Error: Not Found"
+        )
+
+        with pytest.raises(requests.exceptions.HTTPError):
+            pyth_fetch_price(MOCK_PRICE_FEED_ID)

cdp-agentkit-core/python/tests/actions/pyth/test_fetch_price_feed_id.py

@@ -0,0 +1,71 @@
+from unittest.mock import patch
+
+import pytest
+import requests
+
+from cdp_agentkit_core.actions.pyth.fetch_price_feed_id import (
+    PythFetchPriceFeedIDInput,
+    pyth_fetch_price_feed_id,
+)
+
+MOCK_TOKEN_SYMBOL = "BTC"
+
+
+def test_pyth_fetch_price_feed_id_input_model_valid():
+    """Test that PythFetchPriceFeedIDInput accepts valid parameters."""
+    input_model = PythFetchPriceFeedIDInput(
+        token_symbol=MOCK_TOKEN_SYMBOL,
+    )
+
+    assert input_model.token_symbol == MOCK_TOKEN_SYMBOL
+
+
+def test_pyth_fetch_price_feed_id_input_model_missing_params():
+    """Test that PythFetchPriceFeedIDInput raises error when params are missing."""
+    with pytest.raises(ValueError):
+        PythFetchPriceFeedIDInput()
+
+
+def test_pyth_fetch_price_feed_id_success():
+    """Test successful pyth fetch price feed id with valid parameters."""
+    mock_response = {
+        "data": [
+            {
+                "id": "0ff1e87c65eb6e6f7768e66543859b7f3076ba8a3529636f6b2664f367c3344a",
+                "type": "price_feed",
+                "attributes": {"base": "BTC", "quote": "USD", "asset_type": "crypto"},
+            }
+        ]
+    }
+
+    with patch("requests.get") as mock_get:
+        mock_get.return_value.json.return_value = mock_response["data"]
+        mock_get.return_value.raise_for_status.return_value = None
+
+        result = pyth_fetch_price_feed_id(MOCK_TOKEN_SYMBOL)
+
+        assert result == "0ff1e87c65eb6e6f7768e66543859b7f3076ba8a3529636f6b2664f367c3344a"
+        mock_get.assert_called_once_with(
+            "https://hermes.pyth.network/v2/price_feeds?query=BTC&asset_type=crypto"
+        )
+
+
+def test_pyth_fetch_price_feed_id_empty_response():
+    """Test pyth fetch price feed id error with empty response for ticker symbol."""
+    with patch("requests.get") as mock_get:
+        mock_get.return_value.json.return_value = []
+        mock_get.return_value.raise_for_status.return_value = None
+
+        with pytest.raises(ValueError, match="No price feed found for TEST"):
+            pyth_fetch_price_feed_id("TEST")
+
+
+def test_pyth_fetch_price_feed_id_http_error():
+    """Test pyth fetch price feed id error with HTTP error."""
+    with patch("requests.get") as mock_get:
+        mock_get.return_value.raise_for_status.side_effect = requests.exceptions.HTTPError(
+            "404 Client Error: Not Found"
+        )
+
+        with pytest.raises(requests.exceptions.HTTPError):
+            pyth_fetch_price_feed_id(MOCK_TOKEN_SYMBOL)

cdp-agentkit-core/typescript/CHANGELOG.md

@@ -6,6 +6,8 @@
 
 - Added `get_balance_nft` action.
 - Added `transfer_nft` action.
+- Added `pyth_fetch_price_feed_id` action to fetch the price feed ID for a given token symbol from Pyth.
+- Added `pyth_fetch_price` action to fetch the price of a given price feed from Pyth.
 
 ## [0.0.11] - 2025-01-13
 

cdp-agentkit-core/typescript/src/actions/cdp/data/pyth/fetch_price.ts

@@ -0,0 +1,70 @@
+import { CdpAction } from "../../cdp_action";
+import { z } from "zod";
+
+const PYTH_FETCH_PRICE_PROMPT = `
+Fetch the price of a given price feed from Pyth.
+
+Inputs:
+- Pyth price feed ID
+
+Important notes:
+- Do not assume that a random ID is a Pyth price feed ID. If you are confused, ask a clarifying question.
+- This action only fetches price inputs from Pyth price feeds. No other source.
+- If you are asked to fetch the price from Pyth for a ticker symbol such as BTC, you must first use the pyth_fetch_price_feed_id
+action to retrieve the price feed ID before invoking the pyth_Fetch_price action
+`;
+
+/**
+ * Input schema for Pyth fetch price action.
+ */
+export const PythFetchPriceInput = z.object({
+  priceFeedID: z.string().describe("The price feed ID to fetch the price for"),
+});
+
+/**
+ * Fetches the price from Pyth given a Pyth price feed ID.
+ *
+ * @param args - The input arguments for the action.
+ * @returns A message containing the price from the given price feed.
+ */
+export async function pythFetchPrice(args: z.infer<typeof PythFetchPriceInput>): Promise<string> {
+  const url = `https://hermes.pyth.network/v2/updates/price/latest?ids[]=${args.priceFeedID}`;
+  const response = await fetch(url);
+
+  if (!response.ok) {
+    throw new Error(`HTTP error! status: ${response.status}`);
+  }
+
+  const data = await response.json();
+  const parsedData = data.parsed;
+
+  if (parsedData.length === 0) {
+    throw new Error(`No price data found for ${args.priceFeedID}`);
+  }
+
+  const priceInfo = parsedData[0].price;
+  const price = BigInt(priceInfo.price);
+  const exponent = priceInfo.expo;
+
+  if (exponent < 0) {
+    const adjustedPrice = price * BigInt(100);
+    const divisor = BigInt(10) ** BigInt(-exponent);
+    const scaledPrice = adjustedPrice / BigInt(divisor);
+    const priceStr = scaledPrice.toString();
+    const formattedPrice = `${priceStr.slice(0, -2)}.${priceStr.slice(-2)}`;
+    return formattedPrice.startsWith(".") ? `0${formattedPrice}` : formattedPrice;
+  }
+
+  const scaledPrice = price / BigInt(10) ** BigInt(exponent);
+  return scaledPrice.toString();
+}
+
+/**
+ * Pyth fetch price action.
+ */
+export class PythFetchPriceAction implements CdpAction<typeof PythFetchPriceInput> {
+  public name = "pyth_fetch_price";
+  public description = PYTH_FETCH_PRICE_PROMPT;
+  public argsSchema = PythFetchPriceInput;
+  public func = pythFetchPrice;
+}