mcp.py 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859
  1. import json
  2. from django.http import JsonResponse, HttpResponse
  3. from django.views.decorators.csrf import csrf_exempt
  4. from chat.mcp.tools import MCPToolHandler
  5. @csrf_exempt
  6. def mcp_view(request):
  7. request_id = None
  8. try:
  9. data = json.loads(request.body)
  10. method = data.get("method")
  11. params = data.get("params", {})
  12. request_id = data.get("id")
  13. if request_id is None:
  14. return HttpResponse(status=204)
  15. auth_header = request.headers.get("Authorization", "").replace("Bearer ", "")
  16. handler = MCPToolHandler(auth_header)
  17. # 路由方法
  18. if method == "initialize":
  19. result = handler.initialize()
  20. elif method == "tools/list":
  21. result = handler.list_tools()
  22. elif method == "tools/call":
  23. result = handler.call_tool(params)
  24. else:
  25. return JsonResponse({
  26. "jsonrpc": "2.0",
  27. "id": request_id,
  28. "error": {
  29. "code": -32601,
  30. "message": f"Method not found: {method}"
  31. }
  32. })
  33. # 成功响应
  34. return JsonResponse({
  35. "jsonrpc": "2.0",
  36. "id": request_id,
  37. "result": result
  38. })
  39. except Exception as e:
  40. return JsonResponse({
  41. "jsonrpc": "2.0",
  42. "id": request_id,
  43. "error": {
  44. "code": -32603,
  45. "message": f"Internal error: {str(e)}"
  46. }
  47. })