"""Tool definitions for Pyth Network MCP Server."""
from mcp.types import Tool
from typing import Any
import json
import pyth_tools
def get_tool_definitions() -> list[Tool]:
"""Get all available Pyth Network tool definitions."""
return [
Tool(
name="get_price_feeds",
description="Search and filter Pyth Network price feeds by symbol or asset type",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Filter results to price feeds whose symbol contains this string (case insensitive)"
},
"asset_type": {
"type": "string",
"description": "Filter by asset type: crypto, equity, fx, metal, or rates (case insensitive)"
}
}
}
),
Tool(
name="get_latest_price_updates",
description="Get the latest price updates for specific Pyth Network price feed IDs",
inputSchema={
"type": "object",
"required": ["ids"],
"properties": {
"ids": {
"type": "array",
"items": {"type": "string"},
"description": "List of price feed IDs to get updates for"
},
"encoding": {
"type": "string",
"enum": ["hex", "base64"],
"description": "Encoding type for binary data (default: hex)"
},
"parsed": {
"type": "boolean",
"description": "Include parsed price update (default: true)"
},
"ignore_invalid_price_ids": {
"type": "boolean",
"description": "Ignore invalid price IDs (default: false)"
},
"include_binary": {
"type": "boolean",
"description": "Include binary proof data (default: false)"
}
}
}
),
Tool(
name="get_price_updates_at_time",
description="Get historical Pyth Network price updates at or after a specific timestamp",
inputSchema={
"type": "object",
"required": ["publish_time", "ids"],
"properties": {
"publish_time": {
"type": "integer",
"description": "Unix timestamp in seconds"
},
"ids": {
"type": "array",
"items": {"type": "string"},
"description": "List of price feed IDs to get updates for"
},
"encoding": {
"type": "string",
"enum": ["hex", "base64"],
"description": "Encoding type for binary data (default: hex)"
},
"parsed": {
"type": "boolean",
"description": "Include parsed price update (default: true)"
},
"ignore_invalid_price_ids": {
"type": "boolean",
"description": "Ignore invalid price IDs (default: false)"
},
"include_binary": {
"type": "boolean",
"description": "Include binary proof data (default: false)"
}
}
}
),
Tool(
name="get_publisher_stake_caps",
description="Get the most recent publisher stake caps data from Pyth Network",
inputSchema={
"type": "object",
"properties": {
"encoding": {
"type": "string",
"enum": ["hex", "base64"],
"description": "Encoding type for binary data (default: hex)"
},
"parsed": {
"type": "boolean",
"description": "Include parsed update data (default: true)"
},
"include_binary": {
"type": "boolean",
"description": "Include binary proof data (default: false)"
}
}
}
),
Tool(
name="get_twap_latest",
description="Get the latest time-weighted average price (TWAP) from Pyth Network with a custom time window",
inputSchema={
"type": "object",
"required": ["window_seconds", "ids"],
"properties": {
"window_seconds": {
"type": "integer",
"minimum": 1,
"maximum": 600,
"description": "Time window in seconds (1-600). Example: 300 for 5-minute TWAP"
},
"ids": {
"type": "array",
"items": {"type": "string"},
"description": "List of price feed IDs to get TWAP for"
},
"encoding": {
"type": "string",
"enum": ["hex", "base64"],
"description": "Encoding type for binary data (default: hex)"
},
"parsed": {
"type": "boolean",
"description": "Include calculated TWAP in parsed field (default: true)"
},
"ignore_invalid_price_ids": {
"type": "boolean",
"description": "Ignore invalid price IDs (default: false)"
},
"include_binary": {
"type": "boolean",
"description": "Include binary proof data (default: false)"
}
}
}
)
]
def handle_tool_call(name: str, arguments: Any) -> dict:
"""Handle tool execution and return result."""
try:
if name == "get_price_feeds":
result = pyth_tools.get_price_feeds(
query=arguments.get("query"),
asset_type=arguments.get("asset_type")
)
elif name == "get_latest_price_updates":
result = pyth_tools.get_latest_price_updates(
ids=arguments["ids"],
encoding=arguments.get("encoding", "hex"),
parsed=arguments.get("parsed", True),
ignore_invalid_price_ids=arguments.get("ignore_invalid_price_ids", False),
include_binary=arguments.get("include_binary", False)
)
elif name == "get_price_updates_at_time":
result = pyth_tools.get_price_updates_at_time(
publish_time=arguments["publish_time"],
ids=arguments["ids"],
encoding=arguments.get("encoding", "hex"),
parsed=arguments.get("parsed", True),
ignore_invalid_price_ids=arguments.get("ignore_invalid_price_ids", False),
include_binary=arguments.get("include_binary", False)
)
elif name == "get_publisher_stake_caps":
result = pyth_tools.get_publisher_stake_caps(
encoding=arguments.get("encoding", "hex"),
parsed=arguments.get("parsed", True),
include_binary=arguments.get("include_binary", False)
)
elif name == "get_twap_latest":
result = pyth_tools.get_twap_latest(
window_seconds=arguments["window_seconds"],
ids=arguments["ids"],
encoding=arguments.get("encoding", "hex"),
parsed=arguments.get("parsed", True),
ignore_invalid_price_ids=arguments.get("ignore_invalid_price_ids", False),
include_binary=arguments.get("include_binary", False)
)
else:
result = {"error": f"Unknown tool: {name}"}
return result
except Exception as e:
return {"error": f"Error calling {name}: {str(e)}"}