pandaserver/pandamcp/mcp_utils.py

0001 import inspect
0002 import os.path
0003 from collections.abc import Callable
0004
0005 from fastmcp.server.dependencies import get_http_headers
0006 from fastmcp.class="varlink" href="/lxr/search?_filestring=tools">tools.tool import Tool
0007
0008 from pandaserver.api.v1.http_client import HttpClient, api_url, api_url_ssl
0009 from pandaserver.srvcore.panda_request import PandaRequest
0010
0011
0012 def create_tool(func: Callable) -> Tool:
0013     """
0014     Create an MCP tool that wraps the API call.
0015
0016     :param func: The function to wrap. It should be a callable that takes PandaRequest as the first argument.
0017     :return: An MCP tool.
0018     """
0019
0020     # construct the URL based on the module and function name
0021     mod_path = inspect.getfile(inspect.getmodule(func))
0022     mod_name = os.path.basename(mod_path).split("_")[0]
0023
0024     # determine http method based on the docstring
0025     http_method = None
0026     for line in func.__doc__.splitlines():
0027         line = line.strip()
0028         if line.startswith("HTTP Method:"):
0029             http_method = line.split(":")[1].strip().lower()
0030             if http_method not in ["get", "post"]:
0031                 raise ValueError(f"Unsupported HTTP method: {http_method}")
0032             break
0033     if http_method is None:
0034         raise ValueError("HTTP Method not specified in the function docstring")
0035
0036     if http_method == "get":
0037         url = f"{api_url}/{mod_name}/{func.__name__}"
0038     else:
0039         url = f"{api_url_ssl}/{mod_name}/{func.__name__}"
0040
0041     # remove PandaRequest from the signature and annotations
0042     sig = inspect.signature(func)
0043     params = []
0044     for p in sig.parameters.values():
0045         if p.annotation != PandaRequest:
0046             params.append(p)
0047
0048     annotations = {}
0049     for k, v in func.__annotations__.items():
0050         if v != PandaRequest:
0051             annotations[k] = v
0052
0053     # create a new function that wraps the API call
0054     def wrapped_func(**kwarg):
0055         nonlocal url, http_method
0056         # extract the id_token and auth_vo from the headers
0057         original_headers = get_http_headers()
0058         id_token = original_headers.get("authorization")
0059         if id_token and id_token.startswith("Bearer "):
0060             id_token = original_headers["authorization"].split(" ")[1]
0061         auth_vo = original_headers.get("origin")
0062         oidc = id_token is not None
0063
0064         # set the id_token and auth_vo in the HttpClient
0065         http_client = HttpClient()
0066         http_client.override_oidc(oidc, id_token, auth_vo)
0067         status, output = getattr(http_client, http_method)(url, kwarg)
0068         return output
0069
0070     # set the signature and annotations to the wrapped function to align with the original API call
0071     wrapped_func.__signature__ = sig.replace(parameters=params)
0072     wrapped_func.__annotations__ = annotations
0073
0074     return Tool.from_function(wrapped_func, name=func.__name__, description=func.__doc__)