Skip to content

base_server

Base MCP server class for common functionality.

BaseMCPServer

Bases: ABC

Base class for all MADA MCP servers.

Source code in src/mada_tools/shared/base_server.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class BaseMCPServer(ABC):
    """Base class for all MADA MCP servers."""

    def __init__(self, server_name: str, description: Optional[str] = None):
        """
        Initialize the base MCP server.

        Args:
            server_name: Name of the MCP server
            description: Optional description of the server
        """
        self.server_name = server_name
        self.description = description or f"MCP Server for {server_name}"
        # self.mcp will be initialized in run_with_args after parsing config
        self.mcp = None
        # OAuth configuration (set during run_with_args)
        self.oauth_enabled = False

    def get_env_var(self, var_name: str, default: Optional[str] = None, required: bool = False) -> Optional[str]:
        """
        Get environment variable with optional default and required validation.

        Args:
            var_name: Name of the environment variable
            default: Default value if not set
            required: Whether the variable is required

        Returns:
            The environment variable value

        Raises:
            ValueError: If required variable is not set
        """
        value = os.getenv(var_name, default)
        if required and value is None:
            raise ValueError(f"Required environment variable {var_name} is not set")
        return value

    def parse_args(self) -> argparse.Namespace:
        """Parse command line arguments."""
        parser = argparse.ArgumentParser(description=self.description)
        parser.add_argument("--host", default=None, help="Host to bind to")
        parser.add_argument("--port", type=int, help="Port to bind to")
        parser.add_argument("--config", help="Configuration file path")
        parser.add_argument(
            "--transport",
            choices=["stdio", "streamable-http"],
            default="streamable-http",
            help="Transport method (stdio, streamable-http)",
        )
        return parser.parse_args()

    def load_config(self, config_path: str, server_key: str) -> Dict[str, Any]:
        """
        Load configuration from file with environment variable expansion.

        Args:
            config_path: Path to configuration file
            server_key: Key for this server in the config

        Returns:
            Configuration dictionary for this server
        """
        try:
            with open(config_path, "r") as f:
                config = json.load(f)
            return config.get("servers", {}).get(server_key, {})
        except Exception as e:
            print(f"Warning: Could not load config {config_path}: {e}")
            return {}

    def expand_env_vars(self, value: str) -> str:
        """
        Expand environment variable references in configuration values.

        Supports formats:
        - ${VAR_NAME} - expands to os.getenv("VAR_NAME")
        - ${VAR_NAME:-default} - expands with default value if not set

        Args:
            value: String that may contain environment variable references

        Returns:
            String with environment variables expanded
        """
        import re

        def replace_env_var(match):
            var_expr = match.group(1)
            if ":-" in var_expr:
                var_name, default_value = var_expr.split(":-", 1)
                return os.getenv(var_name, default_value)
            else:
                return os.getenv(var_expr, match.group(0))  # Return original if not found

        # Pattern matches ${VAR_NAME} or ${VAR_NAME:-default}
        pattern = r"\$\{([^}]+)\}"
        return re.sub(pattern, replace_env_var, value)

    def run_with_args(self, server_key: str):
        """
        Run the server with command line argument parsing.

        Args:
            server_key: Key for this server in config files
        """
        args = self.parse_args()

        # Load config if provided
        config = {}
        if args.config:
            config = self.load_config(args.config, server_key)
            # Set environment variables from config with expansion
            env_vars = config.get("env_vars", {})
            for key, value in env_vars.items():
                if isinstance(value, str):
                    expanded_value = self.expand_env_vars(value)
                    os.environ.setdefault(key, expanded_value)
                else:
                    os.environ.setdefault(key, str(value))

        # Determine transport method and get host/port
        transport = args.transport or config.get("transport", "streamable-http")

        # Check if OAuth/JWT authentication is enabled in config
        auth_config = config.get("authentication", {})
        self.oauth_enabled = auth_config.get("oauth_enabled", False)

        # Initialize FastMCP with configuration based on transport
        if transport == "stdio":
            # For stdio, host/port are not used
            self.mcp = FastMCP(name=self.server_name)
        else:
            # Configure OAuth/JWT if enabled
            if self.oauth_enabled:
                from fastmcp.server.auth.providers.jwt import JWTVerifier

                jwks_uri = auth_config.get("jwks_uri")
                if not jwks_uri:
                    raise ValueError("OAuth enabled but 'jwks_uri' not specified in authentication config")
                LOG.info("OAuth/JWT authentication enabled")
                LOG.info(f"  JWKS URI: {jwks_uri}")
                verifier = JWTVerifier(
                    jwks_uri=jwks_uri,
                )
                self.mcp = FastMCP(name=self.server_name, auth=verifier)
            else:
                # No authentication
                self.mcp = FastMCP(name=self.server_name)

        # Register tools now that mcp is initialized
        self._register_tools()

        # Start the server
        if transport == "stdio":
            print(f"Starting {self.server_name} with stdio transport")
            self.mcp.run(transport="stdio")
        elif transport == "streamable-http":
            # For HTTP transports, use host/port from config
            host = args.host or config.get("host", "localhost")
            port = args.port or config.get("port", 8000)
            print(f"Starting {self.server_name} with streamable-http on {host}:{port}")
            LOG.info(f"Debug endpoint available at: http://{host}:{port}/debug/headers")
            self.mcp.run(
                transport="streamable-http",
                host=host,
                port=port,
                stateless_http=True,
            )
        else:
            raise ValueError(f"Unsupported transport: {transport}")

    def run_tool(self, func: Callable, *args, **kwargs) -> Any:
        """
        Helper function to run a tool and handle errors.

        Args:
            func: The function/method to execute.

        Returns:
            The successful payload returned by the tool.

        Raises:
            ToolExecutionError: If the tool execution fails.
        """
        try:
            success, payload = func(*args, **kwargs)
            if success:
                return payload
            else:
                raise ToolExecutionError(str(payload))
        except Exception as e:
            last = traceback.extract_tb(e.__traceback__)[-1]
            raise ToolExecutionError(f"Tool execution failed at {last.filename}:{last.lineno} in {last.name}: {e}")

__init__(server_name, description=None)

Initialize the base MCP server.

Parameters:

Name Type Description Default
server_name str

Name of the MCP server

 required
description Optional[str]

Optional description of the server

 None
Source code in src/mada_tools/shared/base_server.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, server_name: str, description: Optional[str] = None):
    """
    Initialize the base MCP server.

    Args:
        server_name: Name of the MCP server
        description: Optional description of the server
    """
    self.server_name = server_name
    self.description = description or f"MCP Server for {server_name}"
    # self.mcp will be initialized in run_with_args after parsing config
    self.mcp = None
    # OAuth configuration (set during run_with_args)
    self.oauth_enabled = False

expand_env_vars(value)

Expand environment variable references in configuration values.

Supports formats: - ${VAR_NAME} - expands to os.getenv("VAR_NAME") - ${VAR_NAME:-default} - expands with default value if not set

Parameters:

Name Type Description Default
value str

String that may contain environment variable references

 required

Returns:

Type Description
str

String with environment variables expanded
Source code in src/mada_tools/shared/base_server.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def expand_env_vars(self, value: str) -> str:
    """
    Expand environment variable references in configuration values.

    Supports formats:
    - ${VAR_NAME} - expands to os.getenv("VAR_NAME")
    - ${VAR_NAME:-default} - expands with default value if not set

    Args:
        value: String that may contain environment variable references

    Returns:
        String with environment variables expanded
    """
    import re

    def replace_env_var(match):
        var_expr = match.group(1)
        if ":-" in var_expr:
            var_name, default_value = var_expr.split(":-", 1)
            return os.getenv(var_name, default_value)
        else:
            return os.getenv(var_expr, match.group(0))  # Return original if not found

    # Pattern matches ${VAR_NAME} or ${VAR_NAME:-default}
    pattern = r"\$\{([^}]+)\}"
    return re.sub(pattern, replace_env_var, value)

get_env_var(var_name, default=None, required=False)

Get environment variable with optional default and required validation.

Parameters:

Name Type Description Default
var_name str

Name of the environment variable

 required
default Optional[str]

Default value if not set

 None
required bool

Whether the variable is required

 False

Returns:

Type Description
Optional[str]

The environment variable value

Raises:

Type Description
ValueError

If required variable is not set
Source code in src/mada_tools/shared/base_server.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def get_env_var(self, var_name: str, default: Optional[str] = None, required: bool = False) -> Optional[str]:
    """
    Get environment variable with optional default and required validation.

    Args:
        var_name: Name of the environment variable
        default: Default value if not set
        required: Whether the variable is required

    Returns:
        The environment variable value

    Raises:
        ValueError: If required variable is not set
    """
    value = os.getenv(var_name, default)
    if required and value is None:
        raise ValueError(f"Required environment variable {var_name} is not set")
    return value

load_config(config_path, server_key)

Load configuration from file with environment variable expansion.

Parameters:

Name Type Description Default
config_path str

Path to configuration file

 required
server_key str

Key for this server in the config

 required

Returns:

Type Description
Dict[str, Any]

Configuration dictionary for this server
Source code in src/mada_tools/shared/base_server.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def load_config(self, config_path: str, server_key: str) -> Dict[str, Any]:
    """
    Load configuration from file with environment variable expansion.

    Args:
        config_path: Path to configuration file
        server_key: Key for this server in the config

    Returns:
        Configuration dictionary for this server
    """
    try:
        with open(config_path, "r") as f:
            config = json.load(f)
        return config.get("servers", {}).get(server_key, {})
    except Exception as e:
        print(f"Warning: Could not load config {config_path}: {e}")
        return {}

parse_args()

Parse command line arguments.

Source code in src/mada_tools/shared/base_server.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
def parse_args(self) -> argparse.Namespace:
    """Parse command line arguments."""
    parser = argparse.ArgumentParser(description=self.description)
    parser.add_argument("--host", default=None, help="Host to bind to")
    parser.add_argument("--port", type=int, help="Port to bind to")
    parser.add_argument("--config", help="Configuration file path")
    parser.add_argument(
        "--transport",
        choices=["stdio", "streamable-http"],
        default="streamable-http",
        help="Transport method (stdio, streamable-http)",
    )
    return parser.parse_args()

run_tool(func, *args, **kwargs)

Helper function to run a tool and handle errors.

Parameters:

Name Type Description Default
func Callable

The function/method to execute.

 required

Returns:

Type Description
Any

The successful payload returned by the tool.

Raises:

Type Description
ToolExecutionError

If the tool execution fails.
Source code in src/mada_tools/shared/base_server.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def run_tool(self, func: Callable, *args, **kwargs) -> Any:
    """
    Helper function to run a tool and handle errors.

    Args:
        func: The function/method to execute.

    Returns:
        The successful payload returned by the tool.

    Raises:
        ToolExecutionError: If the tool execution fails.
    """
    try:
        success, payload = func(*args, **kwargs)
        if success:
            return payload
        else:
            raise ToolExecutionError(str(payload))
    except Exception as e:
        last = traceback.extract_tb(e.__traceback__)[-1]
        raise ToolExecutionError(f"Tool execution failed at {last.filename}:{last.lineno} in {last.name}: {e}")

run_with_args(server_key)

Run the server with command line argument parsing.

Parameters:

Name Type Description Default
server_key str

Key for this server in config files

 required
Source code in src/mada_tools/shared/base_server.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def run_with_args(self, server_key: str):
    """
    Run the server with command line argument parsing.

    Args:
        server_key: Key for this server in config files
    """
    args = self.parse_args()

    # Load config if provided
    config = {}
    if args.config:
        config = self.load_config(args.config, server_key)
        # Set environment variables from config with expansion
        env_vars = config.get("env_vars", {})
        for key, value in env_vars.items():
            if isinstance(value, str):
                expanded_value = self.expand_env_vars(value)
                os.environ.setdefault(key, expanded_value)
            else:
                os.environ.setdefault(key, str(value))

    # Determine transport method and get host/port
    transport = args.transport or config.get("transport", "streamable-http")

    # Check if OAuth/JWT authentication is enabled in config
    auth_config = config.get("authentication", {})
    self.oauth_enabled = auth_config.get("oauth_enabled", False)

    # Initialize FastMCP with configuration based on transport
    if transport == "stdio":
        # For stdio, host/port are not used
        self.mcp = FastMCP(name=self.server_name)
    else:
        # Configure OAuth/JWT if enabled
        if self.oauth_enabled:
            from fastmcp.server.auth.providers.jwt import JWTVerifier

            jwks_uri = auth_config.get("jwks_uri")
            if not jwks_uri:
                raise ValueError("OAuth enabled but 'jwks_uri' not specified in authentication config")
            LOG.info("OAuth/JWT authentication enabled")
            LOG.info(f"  JWKS URI: {jwks_uri}")
            verifier = JWTVerifier(
                jwks_uri=jwks_uri,
            )
            self.mcp = FastMCP(name=self.server_name, auth=verifier)
        else:
            # No authentication
            self.mcp = FastMCP(name=self.server_name)

    # Register tools now that mcp is initialized
    self._register_tools()

    # Start the server
    if transport == "stdio":
        print(f"Starting {self.server_name} with stdio transport")
        self.mcp.run(transport="stdio")
    elif transport == "streamable-http":
        # For HTTP transports, use host/port from config
        host = args.host or config.get("host", "localhost")
        port = args.port or config.get("port", 8000)
        print(f"Starting {self.server_name} with streamable-http on {host}:{port}")
        LOG.info(f"Debug endpoint available at: http://{host}:{port}/debug/headers")
        self.mcp.run(
            transport="streamable-http",
            host=host,
            port=port,
            stateless_http=True,
        )
    else:
        raise ValueError(f"Unsupported transport: {transport}")