Twelve Data MCP Server

Instructions

Implementation Reference

• src/mcp_server_twelve_data/server.py:44-83 (handler)

  Core handler logic executed by the thin wrapper for GetForexPairs. Makes HTTP GET request to Twelve Data API endpoint '/forex_pairs' using the provided Pydantic request model, handles errors, and validates response with Pydantic response 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 generated tools including 'GetForexPairs' via register_all_tools function (imported from .tools, which is autogenerated). The registration uses @server.tool(name="GetForexPairs", description=from_openapi).
  register_all_tools(server=server, _call_endpoint=_call_endpoint)
• scripts/generate_requests_models.py:132-237 (schema)

  Generates the Pydantic input schema GetForexPairsRequest from OpenAPI spec for operationId 'GetForexPairs' (endpoint /forex_pairs). Includes parameters like outputsize, apikey, etc.
  def main(): # Load the OpenAPI specification with open(OPENAPI_PATH, "r", encoding="utf-8") as f: spec = json.load(f) components = spec.get("components", {}).get("schemas", {}) request_models: List[str] = [] request_names: set = set() for path, methods in spec.get("paths", {}).items(): for http_method, op in methods.items(): opid = op.get("operationId") if not opid: continue class_name = canonical_class_name(opid, "Request") # Collect parameters from path, query, header, etc. props: dict = {} for param in op.get("parameters", []): name = param["name"] sch = param.get("schema", {"type": "string"}) typ = python_type(sch, components) required = param.get("required", False) desc = param.get("description") or sch.get("description") examples = collect_examples(param, sch) default = sch.get("default", None) props[name] = { "type": typ, "required": required, "description": desc, "examples": examples, "default": default, } # Collect JSON body properties body = op.get("requestBody", {}) \ .get("content", {}) \ .get("application/json", {}) \ .get("schema") if body: body_sch = resolve_schema(body, components) for name, sch in body_sch.get("properties", {}).items(): typ = python_type(sch, components) required = name in body_sch.get("required", []) desc = sch.get("description") examples = collect_examples({}, sch) default = sch.get("default", None) props[name] = { "type": typ, "required": required, "description": desc, "examples": examples, "default": default, } if "outputsize" not in props: props["outputsize"] = { "type": "int", "required": False, "description": ( "Number of data points to retrieve. Supports values in the range from `1` to `5000`. " "Default `10` when no date parameters are set, otherwise set to maximum" ), "examples": [10], "default": 10, } else: props["outputsize"]["default"] = 10 props["outputsize"]["description"] = props["outputsize"]["description"].replace( 'Default `30`', 'Default `10`' ) props["outputsize"]["examples"] = [10] # Add apikey with default="demo" props["apikey"] = { "type": "str", "required": False, "description": "API key", "examples": ["demo"], "default": "demo", } if "interval" in props: props["interval"]["required"] = False props["interval"]["default"] = "1day" # Append plan availability to the description if x-starting-plan is present starting_plan = op.get("x-starting-plan") description = op.get("description", "") if starting_plan: addon = f" Available starting from the `{starting_plan}` plan." description = (description or "") + addon code = gen_class(class_name, props, description) if class_name not in request_names: request_models.append(code) request_names.add(class_name) # Write all generated models to the target file header = ( "from pydantic import BaseModel, Field\n" "from typing import Any, List, Optional\n\n" ) Path(REQUESTS_FILE).write_text(header + "\n\n".join(request_models), encoding="utf-8") print(f"Generated request models: {REQUESTS_FILE}")
• scripts/generate_tools.py:58-68 (helper)

  Generation script for tools.py containing the specific registration and thin handler def GetForexPairs(params: GetForexPairsRequest, ...) -> GetForexPairs200Response: return _call_endpoint("forex_pairs", ...). op='GetForexPairs', key='forex_pairs' from OpenAPI.
  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)
• scripts/split_opnapi_by_groups.py:10-13 (helper)

  Confirms /forex_pairs endpoint is included in the reference_data group of the OpenAPI spec used for tool generation.
  GROUPS = { "reference_data": [ "/stocks", "/forex_pairs",