"""Instrument search and lookup tools for Schwab MCP Server."""
from typing import Any
from ..client import SchwabClient
GET_INSTRUMENTS_SCHEMA = {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Symbol or search term to look up",
},
"projection": {
"type": "string",
"enum": [
"symbol-search",
"symbol-regex",
"desc-search",
"desc-regex",
"search",
"fundamental",
],
"description": (
"Type of search: symbol-search (default), symbol-regex, "
"desc-search, desc-regex, search, or fundamental"
),
"default": "symbol-search",
},
},
"required": ["symbol"],
}
def _parse_instrument(data: dict) -> dict[str, Any]:
"""Parse instrument data from API response."""
return {
"symbol": data.get("symbol"),
"cusip": data.get("cusip"),
"description": data.get("description"),
"exchange": data.get("exchange"),
"asset_type": data.get("assetType"),
}
def _parse_fundamental(data: dict) -> dict[str, Any]:
"""Parse instrument with fundamental data from API response."""
fundamental = data.get("fundamental", {})
return {
"symbol": data.get("symbol"),
"cusip": data.get("cusip"),
"description": data.get("description"),
"exchange": data.get("exchange"),
"asset_type": data.get("assetType"),
"fundamental": {
"pe_ratio": fundamental.get("peRatio"),
"peg_ratio": fundamental.get("pegRatio"),
"pb_ratio": fundamental.get("pbRatio"),
"div_yield": fundamental.get("divYield"),
"div_amount": fundamental.get("divAmount"),
"div_date": fundamental.get("divDate"),
"div_pay_date": fundamental.get("divPayDate"),
"eps": fundamental.get("eps"),
"week_52_high": fundamental.get("week52High"),
"week_52_low": fundamental.get("week52Low"),
"market_cap": fundamental.get("marketCap"),
"avg_volume_10_day": fundamental.get("avg10DaysVolume"),
"avg_volume_1_year": fundamental.get("avg1YearVolume"),
"shares_outstanding": fundamental.get("sharesOutstanding"),
"beta": fundamental.get("beta"),
"next_div_date": fundamental.get("nextDivDate"),
"next_div_pay_date": fundamental.get("nextDivPayDate"),
},
}
async def get_instruments(client: SchwabClient, args: dict) -> dict[str, Any]:
"""
Search for instruments or get instrument details.
Args:
client: SchwabClient instance
args: Tool arguments with 'symbol' and optional 'projection'
Returns:
Instrument search results or details
"""
symbol = args["symbol"]
projection = args.get("projection", "symbol-search")
response = await client.get_instruments(symbol, projection=projection)
# Response is a dict with 'instruments' key containing a list
instruments_list = response.get("instruments", [])
if projection == "fundamental":
instruments = [_parse_fundamental(inst) for inst in instruments_list]
else:
instruments = [_parse_instrument(inst) for inst in instruments_list]
return {
"query": symbol,
"projection": projection,
"count": len(instruments),
"instruments": instruments,
}
