MCPProxy

def get_mcp(self, **settings: Any) -> "FastMCP":
    mcp = FastMCP(name=self._title, **settings)  # newer mcp

    for func in self._registered_funcs:
        try:
            mcp.tool()(func)
        except PydanticInvalidForJsonSchema as e:
            logger.warning("Could not register function %s: %s", func.__name__, e)

    return mcp

def set_security_params(self, security_params: BaseSecurityParameters, name: str | None = None) -> None:
    if name is not None:
        security = self._security.get(name)
        if security is None:
            raise ValueError(f"Security is not set for '{name}'")

        for match_security in security:
            if match_security.accept(security_params):
                break
        else:
            raise ValueError(f"Security parameters {security_params} do not match security {security}")

    self._security_params[name] = security_params

def put(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
    return self._request("put", path, **kwargs)

def get(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
    return self._request("get", path, **kwargs)

def post(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
    return self._request("post", path, **kwargs)

def delete(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
    return self._request("delete", path, **kwargs)

def head(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
    return self._request("head", path, **kwargs)

def patch(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
    return self._request("patch", path, **kwargs)

@classmethod
@require_optional_import(["datamodel_code_generator", "fastapi_code_generator"], "mcp-proxy-gen")
def generate_code(
    cls,
    input_text: str,
    output_dir: Path,
    disable_timestamp: bool = False,
    custom_visitors: list[Path] | None = None,
) -> str:
    if custom_visitors is None:
        custom_visitors = []
    custom_visitors.append(Path(__file__).parent / "security_schema_visitor.py")

    # with patch_get_parameter_type():
    generate_code(
        input_name="openapi.yaml",
        input_text=input_text,
        encoding="utf-8",
        output_dir=output_dir,
        template_dir=cls._get_template_dir() / "client_template",
        disable_timestamp=disable_timestamp,
        custom_visitors=custom_visitors,
        output_model_type=DataModelType.PydanticV2BaseModel,
    )

    main_path = output_dir / "main.py"

    with main_path.open("r") as f:
        main_py_code = f.read()
    # main_py_code = main_py_code.replace("from .models import", "from models import")
    main_py_code = main_py_code.replace("from .models", "from models")
    # Removing "from __future__ import annotations" to avoid ForwardRef issues, should be fixed in fastapi_code_generator
    main_py_code = main_py_code.replace("from __future__ import annotations", "")

    with main_path.open("w") as f:
        f.write(main_py_code)

    return main_path.stem

def set_globals(self, main: ModuleType, suffix: str) -> None:
    xs = {k: v for k, v in main.__dict__.items() if not k.startswith("__")}
    self._globals = {
        k: v for k, v in xs.items() if hasattr(v, "__module__") and v.__module__ in [f"models_{suffix}", "typing"]
    }

@classmethod
@require_optional_import(["yaml"], "mcp-proxy-gen")
def create(
    cls,
    *,
    openapi_specification: str | None = None,
    openapi_url: str | None = None,
    client_source_path: str | None = None,
    servers: list[dict[str, Any]] | None = None,
    rename_functions: bool = False,
    group_functions: bool = False,
    configuration_type: Literal["json", "yaml"] = "json",
) -> "MCPProxy":
    if (openapi_specification is None) == (openapi_url is None):
        raise ValueError("Either openapi_specification or openapi_url should be provided")

    if openapi_specification is None and openapi_url is not None:
        with requests.get(openapi_url, timeout=10) as response:
            response.raise_for_status()
            openapi_specification = response.text

    openapi_parsed = (
        json.loads(openapi_specification) if configuration_type == "json" else yaml.safe_load(openapi_specification)
    )  # type: ignore [arg-type]

    if servers:
        openapi_parsed["servers"] = servers

    yaml_friendly = yaml.safe_dump(openapi_parsed)

    with optional_temp_path(client_source_path) as td:
        suffix = td.name  # noqa F841

        custom_visitors = []

        if rename_functions:
            custom_visitors.append(Path(__file__).parent / "operation_renaming.py")

        if group_functions:
            custom_visitors.append(Path(__file__).parent / "operation_grouping.py")

        main_name = cls.generate_code(  # noqa F841
            input_text=yaml_friendly,  # type: ignore [arg-type]
            output_dir=td,
            custom_visitors=custom_visitors,
        )
        # add td to sys.path
        try:
            sys.path.append(str(td))
            main = importlib.import_module(main_name, package=td.name)  # nosemgrep
        finally:
            sys.path.remove(str(td))

        client: MCPProxy = main.app  # type: ignore [attr-defined]
        client.set_globals(main, suffix=suffix)

        client.dump_configurations(output_dir=td)

        return client

def dump_configurations(self, output_dir: Path) -> None:
    for tag in self._function_group:
        output_file = output_dir / f"mcp_config_{tag}.json"

        functions = [
            registered_function
            for registered_function in self._registered_funcs
            if registered_function.__name__ in self._function_group[tag]
        ]

        self.dump_configuration(output_file, functions)

    self.dump_configuration(output_dir / "mcp_config.json", self._registered_funcs)

def dump_configuration(self, output_file: Path, functions: list[Callable[..., Any]] = None) -> None:
    # Define paths
    template_dir = MCPProxy._get_template_dir() / "config_template"
    template_file = "config.jinja2"

    # Load Jinja environment
    env = Environment(loader=FileSystemLoader(template_dir), trim_blocks=True, lstrip_blocks=True)

    # Load the template
    template = env.get_template(template_file)
    # Prepare context for rendering
    context = {
        "server_url": self._servers[0]["url"],  # single or list depending on your structure
        "authentications": self._get_authentications(),  # list of auth blocks, we will also need to check _security_params
        "operations": [
            {
                "name": op.__name__,
                "description": op._description.replace("\n", " ").replace("\r", "").strip()
                if op._description is not None
                else "",
            }
            for op in functions
        ],
    }

    # Render the template
    rendered_config = template.render(context)

    # Save the output to a file
    with open(output_file, "w") as f:
        f.write(rendered_config)

def load_configuration(self, config_file: str) -> None:
    with Path(config_file).open("r") as f:
        config_data_str = f.read()

    self.load_configuration_from_string(config_data_str)

def load_configuration_from_string(self, config_data_str: str) -> None:
    config_data = json.loads(config_data_str)
    # Load server URL
    self._servers = [{"url": config_data["server"]["url"]}]

    # Load authentication
    for auth in config_data.get("authentication", []):
        security = BaseSecurity.parse_security_parameters(auth)
        self.set_security_params(security)

    operation_names = [op["name"] for op in config_data.get("operations", [])]

    self._registered_funcs = [func for func in self._registered_funcs if func.__name__ in operation_names]

def get_functions(self) -> list[str]:
    raise DeprecationWarning("Use function_names property instead of get_functions method")

def get_function(self, name: str) -> Callable[..., dict[str, Any]]:
    for f in self._registered_funcs:
        if f.__name__ == name:
            return f
    raise ValueError(f"Function {name} not found")

def set_function(self, name: str, func: Callable[..., dict[str, Any]]) -> None:
    for i, f in enumerate(self._registered_funcs):
        if f.__name__ == name:
            self._registered_funcs[i] = func
            return

    raise ValueError(f"Function {name} not found")

def inject_parameters(self, name: str, **kwargs: Any) -> None:
    raise NotImplementedError("Injecting parameters is not implemented yet")