11import json
2+ import logging
23import os
34
45from .manager import AuthManager
910 create_custom_header_auth ,
1011)
1112
13+ logger = logging .getLogger (__name__ )
14+
1215def setup_auth_from_env () -> AuthManager :
1316 auth_manager = AuthManager ()
1417
1518 api_key = os .getenv ("MCP_API_KEY" )
19+ header_name = os .getenv ("MCP_HEADER_NAME" )
20+ prefix = os .getenv ("MCP_PREFIX" )
1621 if api_key :
17- auth_manager .add_auth_provider ("api_key" , create_api_key_auth (api_key ))
22+ auth_manager .add_auth_provider (
23+ "api_key" ,
24+ create_api_key_auth (
25+ api_key ,
26+ header_name if header_name is not None else "Authorization" ,
27+ prefix if prefix is not None else "Bearer" ,
28+ ),
29+ )
1830
1931 username = os .getenv ("MCP_USERNAME" )
2032 password = os .getenv ("MCP_PASSWORD" )
@@ -36,8 +48,8 @@ def setup_auth_from_env() -> AuthManager:
3648 auth_manager .add_auth_provider (
3749 "custom" , create_custom_header_auth (headers )
3850 )
39- except (json .JSONDecodeError , TypeError ):
40- pass
51+ except (json .JSONDecodeError , TypeError ) as exc :
52+ logger . debug ( "Failed to parse MCP_CUSTOM_HEADERS as JSON: %s" , exc )
4153
4254 tool_mapping = os .getenv ("MCP_TOOL_AUTH_MAPPING" )
4355 if tool_mapping :
@@ -48,8 +60,12 @@ def setup_auth_from_env() -> AuthManager:
4860 auth_manager .map_tool_to_auth (
4961 str (tool_name ), str (auth_provider_name )
5062 )
51- except (json .JSONDecodeError , TypeError ):
52- pass
63+ except (json .JSONDecodeError , TypeError ) as exc :
64+ logger .debug ("Failed to parse MCP_TOOL_AUTH_MAPPING as JSON: %s" , exc )
65+
66+ default_provider = os .getenv ("MCP_DEFAULT_AUTH_PROVIDER" )
67+ if default_provider :
68+ auth_manager .set_default_provider (default_provider )
5369
5470 return auth_manager
5571
@@ -64,41 +80,109 @@ def load_auth_config(config_file: str) -> AuthManager:
6480
6581 providers = config .get ("providers" , {})
6682 for name , provider_config in providers .items ():
67- provider_type = provider_config .get ("type" )
68- if provider_type == "api_key" :
69- auth_manager .add_auth_provider (
70- name ,
71- create_api_key_auth (
72- provider_config ["api_key" ],
73- provider_config .get ("header_name" , "Authorization" ),
74- ),
75- )
76- elif provider_type == "basic" :
77- auth_manager .add_auth_provider (
78- name ,
79- create_basic_auth (
80- provider_config ["username" ], provider_config ["password" ]
81- ),
83+ if not isinstance (provider_config , dict ):
84+ raise ValueError (
85+ f"Error configuring auth provider '{ name }
86+ f"expected an object, got { type (provider_config ).__name__ }
8287 )
83- elif provider_type == "oauth" :
84- auth_manager .add_auth_provider (
85- name ,
86- create_oauth_auth (
87- provider_config ["token" ],
88- provider_config .get ("token_type" , "Bearer" ),
89- ),
90- )
91- elif provider_type == "custom" :
92- headers = provider_config .get ("headers" )
93- if not isinstance (headers , dict ):
94- raise ValueError (f"Provider '{ name } )
95- headers_str : dict [str , str ] = {str (k ): str (v ) for k , v in headers .items ()}
96- auth_manager .add_auth_provider (name , create_custom_header_auth (headers_str ))
97- else :
98- raise ValueError (f"Unknown provider type: { provider_type } )
99-
100- tool_mappings = config .get ("tool_mapping" , {})
101- for tool_name , auth_provider_name in tool_mappings .items ():
88+ provider_type = provider_config .get ("type" )
89+
90+ try :
91+ if provider_type == "api_key" :
92+ if "api_key" not in provider_config :
93+ raise ValueError (
94+ f"Provider '{ name }
95+ "required field 'api_key'. Expected: "
96+ "{'type': 'api_key', 'api_key': 'YOUR_API_KEY'}"
97+ )
98+ auth_manager .add_auth_provider (
99+ name ,
100+ create_api_key_auth (
101+ provider_config ["api_key" ],
102+ provider_config .get ("header_name" , "Authorization" ),
103+ provider_config .get ("prefix" , "Bearer" ),
104+ ),
105+ )
106+ elif provider_type == "basic" :
107+ if "username" not in provider_config :
108+ raise ValueError (
109+ f"Provider '{ name }
110+ "required field 'username'. Expected: "
111+ "{'type': 'basic', 'username': 'user', 'password': 'pass'}"
112+ )
113+ if "password" not in provider_config :
114+ raise ValueError (
115+ f"Provider '{ name }
116+ "required field 'password'. Expected: "
117+ "{'type': 'basic', 'username': 'user', 'password': 'pass'}"
118+ )
119+ auth_manager .add_auth_provider (
120+ name ,
121+ create_basic_auth (
122+ provider_config ["username" ], provider_config ["password" ]
123+ ),
124+ )
125+ elif provider_type == "oauth" :
126+ if "token" not in provider_config :
127+ raise ValueError (
128+ f"Provider '{ name }
129+ "required field 'token'. Expected: "
130+ "{'type': 'oauth', 'token': 'YOUR_TOKEN'}"
131+ )
132+ auth_manager .add_auth_provider (
133+ name ,
134+ create_oauth_auth (
135+ provider_config ["token" ],
136+ provider_config .get ("token_type" , "Bearer" ),
137+ ),
138+ )
139+ elif provider_type == "custom" :
140+ headers = provider_config .get ("headers" )
141+ if not headers :
142+ raise ValueError (
143+ f"Provider '{ name }
144+ "required field 'headers'. Expected: "
145+ "{'type': 'custom', 'headers': {'X-Header': 'value'}}"
146+ )
147+ if not isinstance (headers , dict ):
148+ raise ValueError (
149+ f"Provider '{ name }
150+ f"got { type (headers ).__name__ }
151+ )
152+ headers_str : dict [str , str ] = {
153+ str (k ): str (v ) for k , v in headers .items ()
154+ }
155+ auth_manager .add_auth_provider (
156+ name , create_custom_header_auth (headers_str )
157+ )
158+ else :
159+ raise ValueError (
160+ f"Unknown provider type: '{ provider_type } { name }
161+ f"Supported types: api_key, basic, oauth, custom"
162+ )
163+ except (KeyError , ValueError ) as e :
164+ raise ValueError (f"Error configuring auth provider '{ name } { str (e )} )
165+
166+ tool_mappings = config .get ("tool_mapping" )
167+ legacy_tool_mappings = config .get ("tool_mappings" )
168+ if tool_mappings and legacy_tool_mappings :
169+ raise ValueError (
170+ "Both 'tool_mapping' and legacy 'tool_mappings' are defined. "
171+ "Please use only 'tool_mapping'."
172+ )
173+
174+ final_tool_mappings = tool_mappings or legacy_tool_mappings or {}
175+ if final_tool_mappings and not isinstance (final_tool_mappings , dict ):
176+ raise ValueError (
177+ f"'tool_mapping' must be a dict, "
178+ f"got { type (final_tool_mappings ).__name__ }
179+ )
180+
181+ for tool_name , auth_provider_name in final_tool_mappings .items ():
102182 auth_manager .map_tool_to_auth (tool_name , auth_provider_name )
103183
184+ default_provider = config .get ("default_provider" )
185+ if default_provider :
186+ auth_manager .set_default_provider (default_provider )
187+
104188 return auth_manager
0 commit comments