Twelve Data MCP Server

Instructions

Implementation Reference

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

  Core execution logic for the GetPrice tool and all similar endpoint wrappers. Makes authenticated HTTP request to the /price endpoint on TwelveData API and returns parsed response.

  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:86-89 (registration)

  Calls register_all_tools which registers the GetPrice tool among others.

  # we will not publish large vector db, without it server will work in remote mode
  if vector_db_exists():
      register_all_tools(server=server, _call_endpoint=_call_endpoint)
      u_tool = register_u_tool(

• src/mcp_server_twelve_data/server.py:124-125 (registration)

  Alternative registration path for register_all_tools limiting to first N tools.

  register_all_tools(server=server, _call_endpoint=_call_endpoint)
  all_tools = server._tool_manager._tools

• scripts/generate_tools.py:37-69 (helper)

  Generates the tools.py code containing the @server.tool(name='GetPrice') registration and thin wrapper handler function for the /price endpoint.

  def generate_code(ops):
      def fix_case(name: str) -> str:
          return name[0].upper() + name[1:] if name.lower().startswith("advanced") else name
  
      lines = [
          'from mcp.server import FastMCP',
          'from mcp.server.fastmcp import Context',
          ''
      ]
  
      # Import request models
      for op, _, _ in ops:
          lines.append(f'from .request_models import {fix_case(op)}Request')
      lines.append('')
  
      # Import response models
      for op, _, _ in ops:
          lines.append(f'from .response_models import {fix_case(op)}200Response')
      lines.append('')
  
      # Register tools
      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/generate_tools.py:92-95 (schema)

  The generate_tools.py script processes OpenAPI spec to create tool registrations and relies on request/response models for GetPrice schema validation.

  if __name__ == '__main__':
      main()