chore: refactor MCP configuration retrieval and add MCP view

Author: liuruibin

apps/application/chat_pipeline/step/chat_step/impl/base_chat_step.py

@@ -276,8 +276,8 @@ def _handle_mcp_request(self, mcp_enable, tool_enable, mcp_source, mcp_servers,
                     else:
                         continue
                     executor = ToolExecutor()
-                    app_config = executor.get_app_mcp_config(api_key, app.name, app.desc)
-                    mcp_servers_config[str(app.id)] = app_config
+                    app_config = executor.get_app_mcp_config(api_key)
+                    mcp_servers_config[app.name] = app_config
 
         if len(mcp_servers_config) > 0:
             return mcp_response_generator(chat_model, message_list, json.dumps(mcp_servers_config), mcp_output_enable)

apps/application/flow/step_node/ai_chat_step_node/impl/base_chat_node.py

@@ -263,7 +263,7 @@ def _handle_mcp_request(self, mcp_enable, tool_enable, mcp_source, mcp_servers,
                     else:
                         continue
                     executor = ToolExecutor()
-                    app_config = executor.get_app_mcp_config(api_key, app.name, app.desc)
+                    app_config = executor.get_app_mcp_config(api_key)
                     mcp_servers_config[str(app.id)] = app_config
 
         if len(mcp_servers_config) > 0:

apps/chat/mcp/__init__.py

@@ -0,0 +1,82 @@
+import json
+
+import uuid_utils.compat as uuid
+from django.db.models import QuerySet
+
+from application.models import ApplicationApiKey, Application, ChatUserType
+from chat.serializers.chat import ChatSerializers
+
+
+class MCPToolHandler:
+    def __init__(self, auth_header):
+        app_key = QuerySet(ApplicationApiKey).filter(secret_key=auth_header, is_active=True).first()
+        if not app_key:
+            raise PermissionError("Invalid API Key")
+
+        self.application = QuerySet(Application).filter(id=app_key.application_id, is_publish=True).first()
+        if not self.application:
+            raise PermissionError("Application is not found or not published")
+
+    def initialize(self):
+        return {
+            "protocolVersion": "2025-06-18",
+            "serverInfo": {
+                "name": "maxkb-mcp",
+                "version": "1.0.0"
+            },
+            "capabilities": {
+                "tools": {}
+            }
+        }
+
+    def list_tools(self):
+        return {
+            "tools": [
+                {
+                    "name": 'ai_chat',
+                    "description": f'{self.application.name} {self.application.desc}',
+                    "inputSchema": {
+                        "type": "object",
+                        "properties": {
+                            "message": {"type": "string", "description": "The message to send to the AI."},
+                        },
+                        "required": ["message"]
+                    }
+                }
+            ]
+        }
+
+    def _get_chat_id(self):
+        from application.models import ChatUserType
+        from chat.serializers.chat import OpenChatSerializers
+        from common.init import init_template
+
+        init_template.run()
+
+        return OpenChatSerializers(data={
+            'application_id': self.application.id,
+            'chat_user_id': str(uuid.uuid7()),
+            'chat_user_type': ChatUserType.ANONYMOUS_USER,
+            'debug': False
+        }).open()
+
+    def call_tool(self, params):
+        name = params["name"]
+        args = params.get("arguments", {})
+        # print(params)
+
+        payload = {
+            'message': args.get('message'),
+            'stream': False,
+            're_chat': False
+        }
+        resp = ChatSerializers(data={
+            'chat_id': self._get_chat_id(),
+            'chat_user_id': str(uuid.uuid7()),
+            'chat_user_type': ChatUserType.ANONYMOUS_USER,
+            'application_id': self.application.id,
+            'debug': False,
+        }).chat(payload)
+        data = json.loads(str(resp.text))
+
+        return {"content": [{"type": "text", "text": data.get('data', {}).get('content')}]}

apps/chat/mcp/tools.py

@@ -1,11 +1,13 @@
 from django.urls import path
 
+from chat.views.mcp import mcp_view
 from . import views
 
 app_name = 'chat'
 # @formatter:off
 urlpatterns = [
     path('embed', views.ChatEmbedView.as_view()),
+    path('mcp', mcp_view),
     path('auth/anonymous', views.AnonymousAuthentication.as_view()),
     path('profile', views.AuthProfile.as_view()),
     path('resource_proxy',views.ResourceProxy.as_view()),
@@ -15,8 +17,7 @@
     path('text_to_speech', views.TextToSpeech.as_view()),
     path('speech_to_text', views.SpeechToText.as_view()),
     path('captcha', views.CaptchaView.as_view(), name='captcha'),
-    path('<str:application_id>/chat/completions', views.OpenAIView.as_view(),
-         name='application/chat_completions'),
+    path('<str:application_id>/chat/completions', views.OpenAIView.as_view(), name='application/chat_completions'),
     path('vote/chat/<str:chat_id>/chat_record/<str:chat_record_id>', views.VoteView.as_view(), name='vote'),
     path('historical_conversation', views.HistoricalConversationView.as_view(), name='historical_conversation'),
     path('historical_conversation/<str:chat_id>/record/<str:chat_record_id>',views.ChatRecordView.as_view(),name='conversation_details'),

apps/chat/urls.py

@@ -0,0 +1,59 @@
+import json
+
+from django.http import JsonResponse, HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+
+from chat.mcp.tools import MCPToolHandler
+
+
+@csrf_exempt
+def mcp_view(request):
+    request_id = None
+    try:
+        data = json.loads(request.body)
+        method = data.get("method")
+        params = data.get("params", {})
+        request_id = data.get("id")
+
+        if request_id is None:
+            return HttpResponse(status=204)
+
+        auth_header = request.headers.get("Authorization", "").replace("Bearer ", "")
+        handler = MCPToolHandler(auth_header)
+
+        # 路由方法
+        if method == "initialize":
+            result = handler.initialize()
+
+        elif method == "tools/list":
+            result = handler.list_tools()
+
+        elif method == "tools/call":
+            result = handler.call_tool(params)
+
+        else:
+            return JsonResponse({
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "error": {
+                    "code": -32601,
+                    "message": f"Method not found: {method}"
+                }
+            })
+
+        # 成功响应
+        return JsonResponse({
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": result
+        })
+
+    except Exception as e:
+        return JsonResponse({
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "error": {
+                "code": -32603,
+                "message": f"Internal error: {str(e)}"
+            }
+        })

apps/chat/views/mcp.py

@@ -235,82 +235,13 @@ def get_tool_mcp_config(self, code, params, name, description):
         }
         return tool_config
 
-    def get_app_mcp_config(self, api_key, name, description):
-        chat_path = CONFIG.get_chat_path()
-        # 生成内部令牌(基于时间戳+密钥+api_key)
-        timestamp = int(time.time())
-        secret = CONFIG.get('MCP_INTERNAL_SECRET', 'your-secret-key')
-        token_data = f"{api_key}:{timestamp}"
-        internal_token = hmac.new(
-            secret.encode(),
-            token_data.encode(),
-            hashlib.sha256
-        ).hexdigest()
-        _code = f'''
-from typing import Optional
-
-def _get_chat_id() -> Optional[str]:
-    import requests
-
-    url = f"http://127.0.0.1:8080{chat_path}/api/open"
-    headers = {{
-        'accept': '*/*',
-        'Authorization': f'Bearer {api_key}',
-        'X-MCP-Token': '{internal_token}',  # 添加内部令牌
-        'X-MCP-Timestamp': '{timestamp}'
-    }}
-    try:
-        resp = requests.get(url, headers=headers, timeout=10)
-        resp.raise_for_status()
-        return resp.json().get("data")
-    except Exception as e:
-        raise e
-
-
-def _chat_with_ai(chat_id: str, message: str) -> Optional[str]:
-    import requests
-    
-    url = f"http://127.0.0.1:8080{chat_path}/api/chat_message/{{chat_id}}"
-    headers = {{
-        'Content-Type': 'application/json',
-        'Authorization': f'Bearer {api_key}',
-        'X-MCP-Token': '{internal_token}',  # 添加内部令牌
-        'X-MCP-Timestamp': '{timestamp}'
-    }}
-    payload = {{
-        "message": message,
-        "re_chat": False,
-        "stream": False
-    }}
-    try:
-        resp = requests.post(url, json=payload, headers=headers, timeout=600)
-        resp.raise_for_status()
-        data = resp.json()
-        return str(data.get("data", {{}}).get("content") or data.get("response"))
-    except Exception as e:
-        raise e
-
-def ai_chat(message: str) -> str:
-    chat_id = _get_chat_id()
-    reply = _chat_with_ai(chat_id, message)
-    return reply or "AI 未能生成回复"
-
-        '''
-        _code = self.generate_mcp_server_code(_code, {}, name, description)
-        # print(_code)
-        maxkb_logger.debug(f"Python code of mcp app: {_code}")
-        compressed_and_base64_encoded_code_str = base64.b64encode(gzip.compress(_code.encode())).decode()
+    def get_app_mcp_config(self, api_key):
         app_config = {
-            'command': sys.executable,
-            'args': [
-                '-c',
-                f'import base64,gzip; exec(gzip.decompress(base64.b64decode(\'{compressed_and_base64_encoded_code_str}\')).decode())',
-            ],
-            'cwd': _sandbox_path,
-            'env': {
-                'LD_PRELOAD': f'{_sandbox_path}/lib/sandbox.so',
+            'url': f'http://127.0.0.1:8080{CONFIG.get_chat_path()}/api/mcp',
+            'transport': 'streamable_http',
+            'headers': {
+                'Authorization': f'Bearer {api_key}',
             },
-            'transport': 'stdio',
         }
         return app_config