GetEtf
Retrieve ETF data from Twelve Data API, including daily updated listings with filtering by symbol, exchange, or identifier.
Instructions
This API call returns an array of ETFs available at Twelve Data API. This list is updated daily.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes |
Implementation Reference
- Core handler logic executed by the GetEtf tool (and all autogenerated tools). It resolves path parameters, adds API key, makes HTTP GET request to the Twelve Data API endpoint (e.g., /etf), handles errors, and validates response using Pydantic model.async def _call_endpoint( endpoint: str, params: P, response_model: Type[R], ctx: Context ) -> R: params.apikey = extract_twelve_data_apikey( twelve_data_apikey=twelve_data_apikey, transport=transport, ctx=ctx ) params_dict = params.model_dump(exclude_none=True) resolved_endpoint = resolve_path_params(endpoint, params_dict) async with httpx.AsyncClient( trust_env=False, headers={ "accept": "application/json", "user-agent": "python-httpx/0.24.0" }, ) as client: resp = await client.get( f"{api_base}/{resolved_endpoint}", params=params_dict ) resp.raise_for_status() resp_json = resp.json() if isinstance(resp_json, dict): status = resp_json.get("status") if status == "error": code = resp_json.get('code') raise HTTPException( status_code=code, detail=f"Failed to perform request," f" code = {code}, message = {resp_json.get('message')}" ) return response_model.model_validate(resp_json)
- src/mcp_server_twelve_data/server.py:88-88 (registration)Registers all autogenerated tools including 'GetEtf' using the register_all_tools function, passing the shared _call_endpoint handler.register_all_tools(server=server, _call_endpoint=_call_endpoint)
- Helper function used by the handler to resolve dynamic path parameters in API endpoints (e.g., for symbol-specific ETF queries).def resolve_path_params(endpoint: str, params_dict: dict) -> str:
- src/mcp_server_twelve_data/server.py:124-124 (registration)Alternative registration of all tools when no u_tool OAuth or key is provided, limiting to number_of_tools.register_all_tools(server=server, _call_endpoint=_call_endpoint)
- scripts/generate_tools.py:58-95 (helper)Generation script for tools.py or tools_autogen.py, which defines the specific thin wrapper handler and registration for 'GetEtf' tool, mapping to the /etf endpoint.lines.append('def register_all_tools(server: FastMCP, _call_endpoint):') for op, desc, key in ops: fixed_op = fix_case(op) lines += [ f' @server.tool(name="{op}",', f' description="{desc}")', f' async def {op}(params: {fixed_op}Request, ctx: Context) -> {fixed_op}200Response:', f' return await _call_endpoint("{key}", params, {fixed_op}200Response, ctx)', '' ] return '\n'.join(lines) def main(): spec = load_openapi_spec(OPENAPI_PATH) csv_paths = load_csv_paths(ENDPOINTS_PATH) all_spec_paths = list(spec.get("paths", {}).keys()) extra_paths = sorted(set(all_spec_paths) - set(csv_paths)) final_paths = csv_paths + extra_paths ops = collect_operations(final_paths, spec) total = len(ops) from_csv = len([op for op in ops if '/' + op[2] in csv_paths]) from_extra = total - from_csv print(f"[INFO] Loaded {len(csv_paths)} paths from CSV.") print(f"[INFO] Found {len(all_spec_paths)} paths in OpenAPI spec.") print(f"[INFO] Added {from_extra} additional paths not listed in CSV.") print(f"[INFO] Generated {total} tools in total.") code = '# AUTOGENERATED FILE - DO NOT EDIT MANUALLY\n\n' + generate_code(ops) Path(OUTPUT_PATH).write_text(code, encoding='utf-8') print(f"[SUCCESS] File written to: {OUTPUT_PATH}") if __name__ == '__main__': main()