Twelve Data MCP Server

from typing import Type, TypeVar, Literal, Optional import httpx from pydantic import BaseModel from mcp.server.fastmcp import FastMCP, Context from starlette.exceptions import HTTPException from starlette.requests import Request from starlette.responses import JSONResponse import re from .common import vector_db_exists from .doc_tool import register_doc_tool, register_http_doctool from .doc_tool_remote import register_doc_tool_remote from .key_provider import extract_twelve_data_apikey from .tools import register_all_tools from .u_tool import register_u_tool, register_http_utool from .u_tool_remote import register_u_tool_remote def serve( api_base: str, transport: Literal["stdio", "sse", "streamable-http"], twelve_data_apikey: Optional[str], number_of_tools: int, u_tool_open_ai_api_key: Optional[str], u_tool_oauth2: bool ) -> None: server = FastMCP( "mcp-twelve-data", host="0.0.0.0", port="8000", ) P = TypeVar('P', bound=BaseModel) R = TypeVar('R', bound=BaseModel) def resolve_path_params(endpoint: str, params_dict: dict) -> str: def replacer(match): key = match.group(1) if key not in params_dict: raise ValueError(f"Missing path parameter: {key}") return str(params_dict.pop(key)) return re.sub(r"{(\w+)}", replacer, endpoint) 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) if u_tool_oauth2 or u_tool_open_ai_api_key is not None: # 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( server=server, open_ai_api_key_from_args=u_tool_open_ai_api_key, transport=transport ) doc_tool = register_doc_tool( server=server, open_ai_api_key_from_args=u_tool_open_ai_api_key, transport=transport ) else: u_tool = register_u_tool_remote( server=server, twelve_data_apikey=twelve_data_apikey, open_ai_api_key_from_args=u_tool_open_ai_api_key, transport=transport, ) doc_tool = register_doc_tool_remote( server=server, twelve_data_apikey=twelve_data_apikey, open_ai_api_key_from_args=u_tool_open_ai_api_key, transport=transport, ) register_http_utool( transport=transport, u_tool=u_tool, server=server, ) register_http_doctool( transport=transport, server=server, doc_tool=doc_tool, ) else: register_all_tools(server=server, _call_endpoint=_call_endpoint) all_tools = server._tool_manager._tools server._tool_manager._tools = dict(list(all_tools.items())[:number_of_tools]) @server.custom_route("/health", ["GET"]) async def health(_: Request): return JSONResponse({"status": "ok"}) server.run(transport=transport)