mitmproxy-mcp MCP Server

Instructions

Implementation Reference

• src/mitmproxy_mcp/server.py:163-261 (handler)

  The core handler function for the 'get_flow_details' tool. It retrieves specific flows by index from a mitmproxy session dump, parses JSON content with size-based truncation or structure previews for large payloads, compiles detailed information including headers and content, and returns a JSON-formatted list of flow details.

  async def get_flow_details(arguments: dict) -> list[types.TextContent]:
      """
      Gets details of specific flows from a mitmproxy dump file.
      For large JSON content, returns structure preview instead of full content.
      """
      session_id = arguments.get("session_id")
      flow_indexes = arguments.get("flow_indexes")
      include_content = arguments.get("include_content", True)
  
      if not session_id:
          return [types.TextContent(type="text", text="Error: Missing session_id")]
      if not flow_indexes:
          return [types.TextContent(type="text", text="Error: Missing flow_indexes")]
  
      try:
          flows = await get_flows_from_dump(session_id)
          flow_details_list = []
  
          for flow_index in flow_indexes:
              try:
                  flow = flows[flow_index]
  
                  if flow.type == "http":
                      request = flow.request
                      response = flow.response
  
                      # Parse content
                      request_content = parse_json_content(request.content, dict(request.headers))
                      response_content = None
                      if response:
                          response_content = parse_json_content(response.content, dict(response.headers))
                      
                      # Handle large content
                      request_content_preview = None
                      response_content_preview = None
  
                      flow_details = {}
                      
                      # Check if request content is large and is JSON
                      if include_content and len(request.content) > MAX_CONTENT_SIZE and isinstance(request_content, dict):
                          request_content_preview = generate_json_structure(request_content)
                          request_content = None  # Don't include full content
                      elif include_content and len(request.content) > MAX_CONTENT_SIZE:
                          if isinstance(request_content, str):
                              request_content = request_content[:MAX_CONTENT_SIZE] + " ...[truncated]"
                          else:
                              request_content = request_content[:MAX_CONTENT_SIZE].decode(errors="ignore") + " ...[truncated]"
                          flow_details["request_content_note"] = f"Request content truncated to {MAX_CONTENT_SIZE} bytes."
                      
                      # Check if response content is large and is JSON
                      if response and include_content and len(response.content) > MAX_CONTENT_SIZE and isinstance(response_content, dict):
                          response_content_preview = generate_json_structure(response_content)
                          response_content = None  # Don't include full content
                      elif response and include_content and len(response.content) > MAX_CONTENT_SIZE:
                          if isinstance(response_content, str):
                              response_content = response_content[:MAX_CONTENT_SIZE] + " ...[truncated]"
                          else:
                              response_content = response_content[:MAX_CONTENT_SIZE].decode(errors="ignore") + " ...[truncated]"
                          flow_details["response_content_note"] = f"Response content truncated to {MAX_CONTENT_SIZE} bytes."
  
                      # Build flow details
                      flow_details.update( {
                          "index": flow_index,
                          "method": request.method,
                          "url": request.url,
                          "request_headers": dict(request.headers),
                          "status": response.status_code if response else None,
                          "response_headers": dict(response.headers) if response else None,
                      })
                      
                      # Add content or previews based on size
                      if include_content:
                          if request_content is not None:
                              flow_details["request_content"] = request_content
                          if request_content_preview is not None:
                              flow_details["request_content_preview"] = request_content_preview
                              flow_details["request_content_size"] = len(request.content)
                              flow_details["request_content_note"] = "Content too large to display. Use extract_json_fields tool to get specific values."
                              
                          if response_content is not None:
                              flow_details["response_content"] = response_content
                          if response_content_preview is not None:
                              flow_details["response_content_preview"] = response_content_preview
                              flow_details["response_content_size"] = len(response.content) if response else 0
                              flow_details["response_content_note"] = "Content too large to display. Use extract_json_fields tool to get specific values."
                      
                      flow_details_list.append(flow_details)
                  else:
                      flow_details_list.append({"error": f"Flow {flow_index} is not an HTTP flow"})
  
              except IndexError:
                  flow_details_list.append({"error": f"Flow index {flow_index} out of range"})
  
          return [types.TextContent(type="text", text=json.dumps(flow_details_list, indent=2))]
  
      except FileNotFoundError:
          return [types.TextContent(type="text", text="Error: Session not found")]
      except Exception as e:
          return [types.TextContent(type="text", text=f"Error reading flow details: {str(e)}")]

• src/mitmproxy_mcp/server.py:52-77 (registration)

  Registration of the 'get_flow_details' tool within the list_tools() function, including name, description, and input JSON Schema defining required session_id and flow_indexes parameters.

  types.Tool(
      name="get_flow_details",
      description="Lists HTTP requests/responses from a mitmproxy capture session, showing method, URL, and status codes",
      inputSchema={
          "type": "object",
          "properties": {
              "session_id": {
                  "type": "string",
                  "description": "The ID of the session"
              },
              "flow_indexes": {
                  "type": "array",
                  "items": {
                      "type": "integer"
                  },
                  "description": "The indexes of the flows"
              },
              "include_content": {
                  "type": "boolean",
                  "description": "Whether to include full content in the response (default: true)",
                  "default": True
              }
          },
          "required": ["session_id", "flow_indexes"]
      }
  ),

• src/mitmproxy_mcp/server.py:55-75 (schema)

  JSON Schema for 'get_flow_details' tool inputs, specifying object with required string session_id, array of integer flow_indexes, and optional boolean include_content.

  inputSchema={
      "type": "object",
      "properties": {
          "session_id": {
              "type": "string",
              "description": "The ID of the session"
          },
          "flow_indexes": {
              "type": "array",
              "items": {
                  "type": "integer"
              },
              "description": "The indexes of the flows"
          },
          "include_content": {
              "type": "boolean",
              "description": "Whether to include full content in the response (default: true)",
              "default": True
          }
      },
      "required": ["session_id", "flow_indexes"]

• src/mitmproxy_mcp/server.py:412-413 (helper)

  Dispatch logic in the generic @server.call_tool() handler that routes 'get_flow_details' calls to the specific handler function.

  elif name == "get_flow_details":
      return await get_flow_details(arguments)